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

Exploring algebraic effects inspired by Effect-TS, ZIO, and the Koka language

The Problem: Python’s Hidden Dependencies and Exceptions

Every developer encounters two questions when reading unfamiliar code:

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

Consider a typical Python function:

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 our two 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.

This problem compounds in real applications with layered architectures:

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. The result: 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:

1. Dep[T] — “I need an instance of type T to run”
2. 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 didn’t emerge from nowhere. It 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 at zio.dev
• Effect-TS (TypeScript) — made effects accessible to the JavaScript ecosystem
• koka (Python) — brings these ideas to Python, leveraging Python 3.13’s type system

This library is an exploration of 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

Let’s build a realistic scenario with multiple failure modes that map to different HTTP responses.

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

Now we compose these layers into a single 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

Note that CacheMiss doesn’t appear in the final signature—we handled it 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. If you forget to .provide(Database()), you get a type error—not a runtime AttributeError.

How yield from Achieves This

The magic happens through Python’s generator protocol. Here’s how it works.

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 receives this request and looks up the handler
3. The runtime sends the Database instance back into the generator
4. db receives the Database instance

The type checker sees:

• You’re yielding Dep[Database] (adds to effect set K)
• You’re receiving Database (the return type)

How Err[E] Works

class Err[E: Exception]:
    def __init__(self, err: E) -> None:
        self.error = err

    def __iter__(self) -> Generator[E, Never, Never]:
        yield self.error  # Yield the error, never resume
        raise RuntimeError("Unreachable")  # Type: Never

When you write return (yield from Err(NotFoundError(...))):

1. The generator yields the NotFoundError exception
2. The runtime receives it and returns it as the final result
3. The generator never resumes (hence Never return type)

The type checker sees:

• You’re yielding NotFoundError (adds to effect set K)
• The expression type is Never (computation doesn’t continue)

Effect Composition

When you yield from authenticate(token) inside get_user_profile:

• authenticate has effects Dep[AuthService] | AuthError
• These get added to get_user_profile‘s effect set
• The type checker automatically computes the union:

  Dep[AuthService] | AuthError | Dep[CacheService] | Dep[Database] | NotFoundError

This is why you don’t need to manually declare the effect types—the type checker infers them from usage.

The Alternative: Manual Return Types

Can we achieve the same type safety without generators? Let’s try using explicit return types.

Attempt: Pass Dependencies, Return Union Types

def authenticate(
    auth: AuthService,
    token: str
) -> User | AuthError:
    user: User | None = auth.verify(token)
    if user is None:
        return AuthError("Invalid token")
    return user


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


def get_user_profile(
    auth: AuthService,
    cache: CacheService,
    db: Database,
    token: str,
    user_id: str
) -> UserProfile | AuthError | NotFoundError:
    # Check auth
    auth_result: User | AuthError = authenticate(auth, token)
    if isinstance(auth_result, AuthError):
        return auth_result
    user: User = auth_result

    # Try cache
    cached: UserProfile | None = cache.get(f"user:{user_id}")
    if cached is not None:
        return cached

    # Try database
    db_result: UserProfile | NotFoundError = get_from_db(db, user_id)
    if isinstance(db_result, NotFoundError):
        return db_result

    return db_result

The Problems

1. Dependency Threading

Every function must explicitly accept and pass through all dependencies. get_user_profile takes 5 parameters—3 of which are just dependency plumbing. In deep call stacks, this becomes unmanageable:

def handle_request(
    auth: AuthService,
    cache: CacheService,
    db: Database,
    email: EmailService,
    logger: Logger,
    metrics: MetricsService,
    token: str,
    user_id: str
) -> ...:
    # Every layer needs all these parameters
    ...

2. Manual Error Type Union

You must manually declare -> UserProfile | AuthError | NotFoundError. Forget one, and callers get type errors—or worse, the type checker misses a case. As you add more error types, the union grows unwieldy:

def complex_operation(...) -> Result | ErrorA | ErrorB | ErrorC | ErrorD | ErrorE:
    ...

3. Nested isinstance Checks Destroy Readability

result1 = step1(...)
if isinstance(result1, Error1):
    return result1

result2 = step2(result1, ...)
if isinstance(result2, Error2):
    return result2

result3 = step3(result2, ...)
if isinstance(result3, Error3):
    return result3

# This continues for EVERY call that can fail

Linear code becomes a staircase of error checks, obscuring the happy path.

4. No Automatic Inference

The type checker can’t infer what dependencies or errors are needed—you must declare everything manually. Add a new error case deep in the call stack? Update every function signature up to the top.

Comparison Table

The yield from Version (For Contrast)

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

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

    return (yield from get_from_db(user_id))

Two parameters. Linear flow. Types inferred automatically.

Why Not async/await?

A natural question: Python already has async/await for suspending and resuming functions. Why not use that?

Different problem domains. async/await solves I/O scheduling—letting the event loop run other tasks while waiting for network or disk. It doesn’t track “what capabilities a function needs.”

async def get_user(user_id: str) -> User:
    ...  # What dependencies? What errors? The signature doesn't say.

There’s no mechanism to declare “this async function needs a Database” in a way the type checker can verify.

Function coloring. Async functions can only be called from async contexts. This “colors” your entire codebase—one async function forces async up the call stack. Generators don’t have this problem; they’re just values you can compose and run synchronously.

No custom interpretation. With async/await, the event loop is the only interpreter. With generators, we can write custom runtimes that handle effects differently—inject mocks for testing, swap implementations for different environments, etc.

An Experiment Worth Exploring

Let’s be clear: koka is experimental. It’s a proof of concept exploring whether Python can express the effect patterns that have proven valuable in TypeScript (Effect-TS), Scala (ZIO), and research languages (Koka).

What makes this possible now?

Python 3.13 introduced new type syntax (PEP 695) that makes generic types more ergonomic. Combined with mature type checkers like pyright, we can finally express complex type relationships that would have been impossibly verbose before.

What’s the value for Python developers?

This library demonstrates that type-checked effect systems are possible in Python—today, with zero runtime dependencies beyond standard generators. It shows a path toward code where:

• Dependencies are explicit and verified at type-check time
• Errors are tracked and exhaustively handled
• Function signatures serve as reliable documentation

This invites a question for the Python community: Should we have better support for this pattern? Could future Python versions make effect tracking even more ergonomic?

Try it yourself:

pip install koka

The goal: answer “what does this function need?” and “what can go wrong?” at a glance. Let your type checker become your documentation.