software

Koka.py：在 Python 中的类型检查依赖注入与错误处理

发布: (2025年12月15日 GMT+8 09:56)
7 min read
原文: Dev.to

Source: Dev.to

问题：Python 中隐藏的依赖和异常

每个开发者在阅读陌生代码时都会遇到两个问题：

  • “这个函数运行需要什么？”
  • “调用它时可能会出什么错？”
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)

要回答这些问题，你必须阅读完整的实现。auth_service 从哪来？verify() 失败时会怎样？find_user() 会返回 None 吗？类型签名 (str) -> UserProfile 几乎没有提供任何信息。

在具有分层架构的真实应用中，这个问题会更加严重：

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

每一层都可能有自己的依赖和失败模式。传统的 Python 将依赖隐式化、错误未检查，导致运行时惊喜、缺失错误处理以及不清晰的契约。

如果类型签名能告诉你所有信息会怎样？

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

现在你 知道：这个函数需要 AuthService、CacheService 和 Database，可能会因 AuthError 或 NotFoundError 失败，并返回 UserProfile——全部无需阅读实现代码。这正是 koka 库所实现的功能。

什么是代数效应？（实用视角）

跳过学术理论，直接看实用的框架：

效应是声明的能力——函数运行所需的东西以及它可能的失败方式。把隐式的东西显式化，并让类型检查器验证它。

koka 库提供了两个关键效应：

  • Dep[T] — “我需要一个类型为 T 的实例来运行”
  • Err[E] — “我可能会因错误 E 失败”

关键洞见：效应会自动组合。如果函数 A 需要 Dep[Database]，函数 B 需要 Dep[Cache]，而函数 C 同时调用 A 与 B，则 C 的类型会自动包含 Dep[Database] | Dep[Cache]。类型检查器会推断这一点——无需手动标注。

灵感来源链

这一模式沿袭了以下创新脉络：

  • 学术研究：代数效应与处理器（Plotkin、Pretnar 等）
  • Koka 语言 — 首个将实用代数效应引入的研究语言
  • ZIO（Scala）— 将效应系统带入主流函数式编程
  • Effect‑TS（TypeScript）— 让效应在 JavaScript 生态中可用
  • koka（Python）— 将这些思想引入 Python，利用 Python 3.13 的类型系统

该库探讨了 Python 的类型系统和生成器是否能够表达在其他生态中已被证明有价值的相同模式。

实际案例：用户查询服务

第 1 层：认证

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

类型签名 Eff[Dep[AuthService] | AuthError, User] 告诉你：

  • 需要：AuthService
  • 可能失败：AuthError
  • 返回：User

第 2 层：缓存查询

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

第 3 层：数据库查询

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

第 4 层：组合处理器

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))

类型签名揭示了一切：

  • 依赖：AuthService、CacheService、Database
  • 可能的错误：AuthError（→ 401）、NotFoundError（→ 404）
  • 返回类型：UserProfile

CacheMiss 未出现在最终签名中，因为它在内部通过回退到数据库得到处理。

运行效应

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())

Koka 运行时确保所有依赖都已提供。忘记 .provide(Database()) 会导致 类型错误，而不是运行时的 AttributeError。

yield from 如何实现这一点

Eff 类型

type Eff[K, R] = Generator[K, Never, R]
  • K — 产生的效应（Dep[T] 类型和异常类型的并集）
  • R — 返回类型

Dep[T] 的工作原理

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

当你写 db = yield from Dep(Database) 时：

  1. 生成器会产出 Dep(Database) —— 请求一个 Database 实例。
  2. 运行时拦截该请求，提供具体实例，并把它送回生成器。

这种模式让类型系统能够在调用图中追踪所需的依赖和可能的错误，提供了与 Koka、ZIO、Effect‑TS 等语言中代数效应系统相同的保证。

Back to Blog

相关文章

阅读更多 »