Spakky Data

Spakky Framework를 위한 데이터 접근 레이어 추상화입니다. Repository와 transaction boundary를 application layer에서 backend-neutral하게 선언하도록 돕습니다.

설치

pip install spakky-data

주요 기능

• Repository 패턴: Aggregate 영속화를 위한 generic repository 인터페이스
• 트랜잭션 관리: autocommit을 지원하는 추상 transaction 클래스
• External Proxy: 데이터베이스가 아닌 외부 service/storage 데이터 접근용 proxy 패턴

spakky-data는 core 추상화 패키지입니다. SQLAlchemy engine/session/table mapping 구현체가 필요하면 spakky-sqlalchemy를 함께 설치하세요.

설계 원칙

Repository는 영속화 전용

Repository는 domain aggregate에 대한 CRUD operation만 처리합니다. find_by_xxx, search_xxx 같은 query method를 repository에 추가하지 마세요.

복잡한 query는 ORM/SQL을 사용해 QueryUseCase에서 직접 구현해야 합니다. Query 관심사를 도메인 레이어 밖에 두어 도메인 오염을 방지합니다.

# ❌ 잘못된 예: repository에 query 관심사 포함
class IUserRepository:
    def find_by_email(self, email: str) -> User | None: ...

# ✅ 올바른 예: QueryUseCase에서 직접 구현
@UseCase()
class FindUserByEmailUseCase(IAsyncQueryUseCase[FindUserByEmailQuery, UserDTO]):
    async def run(self, query: FindUserByEmailQuery) -> UserDTO:
        # ORM/SQL 직접 사용
        ...

외부 연동 Proxy와 Repository

빠른 시작

SQLAlchemy와 함께 쓰는 기본 흐름

사용자 애플리케이션에서 가장 흔한 구성은 spakky-data의 Repository/Transaction 계약을 spakky-sqlalchemy 구현체로 채우는 방식입니다.

pip install spakky-data spakky-sqlalchemy

export SPAKKY_SQLALCHEMY__CONNECTION_STRING="postgresql+psycopg://user:pass@localhost/app"
export SPAKKY_SQLALCHEMY__SUPPORT_ASYNC_MODE="true"

흐름은 다음 순서로 구성합니다.

1. AbstractAggregateRoot 기반 domain aggregate를 정의합니다.
2. spakky.plugins.sqlalchemy.orm.table.AbstractMappableTable과 @Table(Domain)으로 ORM table을 매핑합니다.
3. spakky.plugins.sqlalchemy.persistency.repository.AbstractAsyncGenericRepository를 상속한 @Repository() 구현체를 등록합니다.
4. @UseCase() 메서드에 @Transactional()을 붙이고 Repository를 호출합니다.
5. 복잡한 조회는 Repository에 query method를 추가하지 않고 QueryUseCase에서 AsyncSessionManager 또는 SessionManager로 직접 SQLAlchemy query를 실행합니다.

전체 예제는 데이터베이스 가이드를 참고하세요.

Repository 패턴

domain aggregate용 repository interface를 정의합니다.

from abc import abstractmethod
from uuid import UUID

from spakky.data.persistency.repository import IAsyncGenericRepository
from spakky.domain.models.aggregate_root import AbstractAggregateRoot


class User(AbstractAggregateRoot[UUID]):
    name: str
    email: str


# Repository interface: CRUD only, query method 없음
class IUserRepository(IAsyncGenericRepository[User, UUID]):
    pass  # get, get_or_none, contains, range, save, save_all, delete, delete_all

Transaction 관리

database operation에는 추상 transaction을 사용합니다. 직접 구현해야 하는 경우도 있지만, SQLAlchemy를 쓴다면 spakky-sqlalchemy의 Transaction / AsyncTransaction이 이 계약을 이미 구현합니다.

from spakky.data.persistency.transaction import AbstractAsyncTransaction


class SQLAlchemyTransaction(AbstractAsyncTransaction):
    def __init__(self, session_factory, autocommit: bool = True) -> None:
        super().__init__(autocommit)
        self.session_factory = session_factory
        self.session = None

    async def initialize(self) -> None:
        self.session = self.session_factory()

    async def dispose(self) -> None:
        await self.session.close()

    async def commit(self) -> None:
        await self.session.commit()

    async def rollback(self) -> None:
        await self.session.rollback()

context manager로 사용할 수 있습니다.

async with transaction:
    user = await repository.get(user_id)
    user.name = "New Name"
    await repository.save(user)
    # 성공 시 자동 commit, exception 발생 시 rollback

외부 연동 Proxy 패턴

Proxy 인터페이스로 데이터베이스가 아닌 외부 서비스에 접근합니다. REST API, gRPC 서비스, legacy 시스템 등에 사용합니다.

from spakky.data.external.proxy import ProxyModel, IAsyncGenericProxy


# 외부 payment service의 data model
class PaymentInfo(ProxyModel[str]):
    transaction_id: str
    amount: int
    status: str


# 외부 payment service용 proxy interface
class IPaymentProxy(IAsyncGenericProxy[PaymentInfo, str]):
    pass


# 구현체는 외부 API 호출
class PaymentServiceProxy(IPaymentProxy):
    async def get(self, proxy_id: str) -> PaymentInfo:
        response = await self._http_client.get(f"/payments/{proxy_id}")
        return PaymentInfo(...)

API 레퍼런스

영속성

외부 연동

에러

관련 패키지

개발 검증

패키지 단위 검증은 해당 패키지 디렉토리에서 실행합니다.

uv run ruff format .
uv run ruff check .
uv run pyrefly check
uv run pytest

pytest는 각 패키지 pyproject.toml의 coverage 설정을 사용합니다.

라이선스

MIT License