Koka.py: 파이썬에서 타입 검사된 의존성 주입 및 오류 처리

발행: (2025년 12월 15일 오전 10:56 GMT+9)
8 min read
원문: Dev.to

Source: Dev.to

문제: 파이썬의 숨겨진 의존성 및 예외

모든 개발자는 익숙하지 않은 코드를 읽을 때 두 가지 질문에 직면합니다:

  • “이 함수가 동작하려면 무엇이 필요할까?”
  • “이 함수를 호출하면 어떤 일이 잘못될 수 있을까?”
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()

각 계층은 자체적인 의존성 및 실패 모드를 가질 수 있습니다. 전통적인 파이썬은 의존성을 암묵적으로 두고 오류를 체크하지 않아 런타임에 깜짝 놀라게 되고, 오류 처리가 누락되며 계약이 불분명해집니다.

타입 시그니처가 모든 것을 알려준다면?

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의 타입은 자동으로 Dep[Database] | Dep[Cache]를 포함합니다. 타입 체커가 이를 추론하므로 수동으로 어노테이션을 달 필요가 없습니다.

영감의 흐름

이 패턴은 다음과 같은 혁신의 연속을 따릅니다:

  • 대수 효과와 핸들러에 관한 학술 연구 (Plotkin, Pretnar 등)
  • Koka 언어 — 실용적인 대수 효과를 최초로 도입한 연구 언어
  • ZIO (Scala) — 효과 시스템을 주류 함수형 프로그래밍에 도입
  • Effect‑TS (TypeScript) — 자바스크립트 생태계에 효과를 제공
  • koka (Python) — 파이썬 3.13의 타입 시스템을 활용해 이러한 아이디어를 파이썬에 도입

이 라이브러리는 파이썬의 타입 시스템과 제너레이터가 다른 생태계에서 입증된 패턴을 표현할 수 있는지를 탐구합니다.

실제 예시: 사용자 조회 서비스

레이어 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]
  • Kyield되는 효과들의 집합(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)yield합니다 — 이는 Database 인스턴스를 요청하는 신호입니다.
  2. 런타임이 이 요청을 가로채어 구체적인 인스턴스를 제공하고, 다시 제너레이터에 전달합니다.

이 패턴을 통해 타입 시스템은 호출 그래프 전체에 걸쳐 필요한 의존성과 가능한 오류를 추적할 수 있으며, Koka, ZIO, Effect‑TS와 같은 언어에서 제공하는 대수 효과 시스템과 동일한 보장을 제공합니다.

Back to Blog

관련 글

더 보기 »