Koka.py: Type-Checked Dependency Injection and Error Handling in Python

Source: Dev.to

The Problem: Python’s Hidden Dependencies and Exceptions

Every developer encounters two questions when reading unfamiliar code:

• “What does this function need to work?”
• “What can go wrong when I call this?”

def get_user_profile(user_id: str) -> UserProfile:
    token = request.headers.get("Authorization")
    user = auth_service.verify(token)

    cached = cache.get(f"user:{user_id}")
    if cached:
        return cached

    return database.find_user(user_id)

To answer these questions, you must read the entire implementation. Where does auth_service come from? What happens if verify() fails? Can find_user() return None? The type signature (str) -> UserProfile tells you almost nothing.

In real applications with layered architectures, the problem compounds:

handle_request()
 → get_user_profile()
  → authenticate()
   → check_cache()
    → query_db()

Each layer can have its own dependencies and failure modes. Traditional Python keeps dependencies implicit and errors unchecked, leading to runtime surprises, missing error handling, and unclear contracts.

What if the type signature told you everything?

def get_user_profile(user_id: str) -> Eff[
    Dep[AuthService] | Dep[Database] | Dep[CacheService] | AuthError | NotFoundError,
    UserProfile
]:
    ...

Now you know: this function needs AuthService, CacheService, and Database, might fail with AuthError or NotFoundError, and returns UserProfile—all without reading a single line of implementation. This is what the koka library enables.

What Are Algebraic Effects? (A Practical Take)

Skip the academic theory. Here’s the practical framing:

Effects are declared capabilities—what a function needs to run and how it can fail. Think of it as making the implicit explicit, and letting the type checker verify it.

The koka library provides two key effects:

• Dep[T] — “I need an instance of type T to run”
• Err[E] — “I might fail with error E”

The key insight: effects compose automatically. If function A needs Dep[Database] and function B needs Dep[Cache], and function C calls both A and B, then C’s type automatically includes Dep[Database] | Dep[Cache]. The type checker infers this—no manual annotation needed.

The Inspiration Chain

This pattern follows a lineage of innovation:

• Academic research on algebraic effects and handlers (Plotkin, Pretnar, et al.)
• Koka language — a research language that pioneered practical algebraic effects
• ZIO (Scala) — brought effect systems to mainstream functional programming
• Effect‑TS (TypeScript) — made effects accessible to the JavaScript ecosystem
• koka (Python) — brings these ideas to Python, leveraging Python 3.13’s type system

The library explores whether Python’s type system and generators can express the same patterns that have proven valuable in other ecosystems.

Real‑World Example: User Lookup Service

Layer 1: Authentication

class AuthError(Exception):
    """Invalid or missing authentication. Maps to HTTP 401 Unauthorized."""
    pass

class AuthService:
    def verify(self, token: str) -> User | None:
        ...

def authenticate(token: str) -> Eff[Dep[AuthService] | AuthError, User]:
    auth: AuthService = yield from Dep(AuthService)
    user: User | None = auth.verify(token)
    if user is None:
        return (yield from Err(AuthError("Invalid token")))
    return user

The type signature Eff[Dep[AuthService] | AuthError, User] tells you:

• Needs: AuthService
• Can fail with: AuthError
• Returns: User

Layer 2: Cache Lookup

class CacheMiss(Exception):
    """Internal signal—don't expose to HTTP clients."""
    pass

class CacheService:
    def get(self, key: str) -> UserProfile | None:
        ...

def get_from_cache(user_id: str) -> Eff[Dep[CacheService] | CacheMiss, UserProfile]:
    cache: CacheService = yield from Dep(CacheService)
    data: UserProfile | None = cache.get(f"user:{user_id}")
    if data is None:
        return (yield from Err(CacheMiss()))
    return data

Layer 3: Database Lookup

class NotFoundError(Exception):
    """User doesn't exist. Maps to HTTP 404 Not Found."""
    pass

class Database:
    def find_user(self, user_id: str) -> UserProfile | None:
        ...

def get_from_db(user_id: str) -> Eff[Dep[Database] | NotFoundError, UserProfile]:
    db: Database = yield from Dep(Database)
    data: UserProfile | None = db.find_user(user_id)
    if data is None:
        return (yield from Err(NotFoundError(f"User {user_id} not found")))
    return data

Layer 4: Composed Handler

def get_user_profile(
    token: str,
    user_id: str
) -> Eff[
    Dep[AuthService] | Dep[CacheService] | Dep[Database] | AuthError | NotFoundError,
    UserProfile
]:
    # First authenticate (can fail with AuthError → 401)
    user: User = yield from authenticate(token)

    # Try cache first (handle CacheMiss internally)
    cache: CacheService = yield from Dep(CacheService)
    cached: UserProfile | None = cache.get(f"user:{user_id}")
    if cached is not None:
        return cached

    # Fall back to database (can fail with NotFoundError → 404)
    return (yield from get_from_db(user_id))

The type signature reveals everything:

• Dependencies: AuthService, CacheService, Database
• Possible errors: AuthError (→ 401), NotFoundError (→ 404)
• Return type: UserProfile

CacheMiss does not appear in the final signature because it is handled internally by falling back to the database.

Running the Effect

result: UserProfile | AuthError | NotFoundError = (
    Koka()
    .provide(AuthService())
    .provide(CacheService())
    .provide(Database())
    .run(get_user_profile("token-123", "user-456"))
)

# Pattern match to HTTP responses
match result:
    case AuthError() as e:
        return HttpResponse(401, f"Unauthorized: {e}")
    case NotFoundError() as e:
        return HttpResponse(404, f"Not Found: {e}")
    case UserProfile() as profile:
        return HttpResponse(200, profile.to_json())

The Koka runtime ensures all dependencies are provided. Forgetting to .provide(Database()) results in a type error, not a runtime AttributeError.

How yield from Achieves This

The Eff Type

type Eff[K, R] = Generator[K, Never, R]

• K — the effects yielded (union of Dep[T] types and exception types)
• R — the return type

How Dep[T] Works

class Dep[T]:
    def __init__(self, tpe: type[T]) -> None:
        self.tpe = tpe

    def __iter__(self) -> Generator[Self, T, T]:
        return (yield self)  # Yield the request, receive the instance

When you write db = yield from Dep(Database):

1. The generator yields Dep(Database) — a request for a Database instance.
2. The runtime intercepts the request, supplies the concrete instance, and sends it back into the generator.

This pattern lets the type system track required dependencies and possible errors throughout the call graph, providing the same guarantees that algebraic‑effect systems offer in languages like Koka, ZIO, and Effect‑TS.