Koka.py: 파이썬에서 타입 검사된 의존성 주입 및 오류 처리
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]
K—yield되는 효과들의 집합(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)를 작성하면:
- 제너레이터가
Dep(Database)를yield합니다 — 이는Database인스턴스를 요청하는 신호입니다. - 런타임이 이 요청을 가로채어 구체적인 인스턴스를 제공하고, 다시 제너레이터에 전달합니다.
이 패턴을 통해 타입 시스템은 호출 그래프 전체에 걸쳐 필요한 의존성과 가능한 오류를 추적할 수 있으며, Koka, ZIO, Effect‑TS와 같은 언어에서 제공하는 대수 효과 시스템과 동일한 보장을 제공합니다.