파이썬과 DDD로 라이프 캘린더 만들기 #4 레파지토리

이전 글에서 영구 저장에 필요한 데이터베이스를 실제로 준비했습니다. 하지만 아직은 Calendar 클래스를 통해 만들어진 인스턴스를 위에서 설정한 데이터베이스에 저장할 방법이 없습니다. 도메인 모델을 작성할 때는 일부러 무시했던 '지속성'의 역할을 담당할 객체가 필요합니다. DDD에서는 이런 역할을 '레파지토리'라는 객체에게 맡깁니다. '레파지토리 패턴'에 대한 자세한 내용은 이 문서를 참고하시면 좋을 것 같습니다.

구현은 기본적으로 이 문서를 참고했으나, 저는 문서와는 살짝 다르게 데이터베이스 컨텍스트와 레파지토리를 분리해서 사용하게 따로 구현했습니다.

데이터베이스 컨텍스트

우선 컨텍스트 인터페이스를 선언해 주어야 합니다. 컨텍스트의 인터페이스를 굳이 만드는 이유는, 앞으로 이 컨텍스트를 사용할 '응용 계층'의 테스트 코드를 작성할 때, 실제 데이터베이스 인스턴스를 실행하고, 스키마를 설정하고, 데이터 row을 INSERT -> DELETE 하지 않아도 되게 만들기 위해서입니다. 컨텍스트를 인터페이스로 설정하면 테스트에서는 이를 mock 객체로 쉽게 교체할 수 있습니다. 그러면 번거로운 데이터베이스 및 인프라 설정 과정 없이, 빠르게 실행되는 테스트, 원하는 타겟 코드에 집중하는 테스트를 작성할 수 있습니다. 자세한 내용은 '응용 계층'을 작성할 때 다시 이야기해 보겠습니다.

이 인터페이스에는 PostgreSQL이나 SQLAlchemy 같은 구체적 기술이 들어가있지 않으므로 도메인 계층 src/domain/repo/i_db_context.py 경로에 작성해 줍니다.

import abc
import contextlib
from typing import Iterator


class IDBContext(abc.ABC):
    @contextlib.contextmanager
    @abc.abstractmethod
    def begin_transaction(self) -> Iterator[None]:
        ...

저는 트랜잭션의 범위를 명시적으로 관리하는 것을 좋아합니다(쿼리가 나갈 때 마다 자동으로 트랜잭션이 열리고 닫히는 것 보다는). 파이썬의 with 절을 사용하면 트랜잭션의 단위가 더욱 명시적으로 눈에 띄게할 수 있습니다. 이를 위해 contextlib을 사용합니다.

이제 IDBContext 인터페이스를 만족하는 SQLAlchemy 구현체를 작성해 봅시다. SQLAlchemy는 Session 객체를 통해 트랜잭션 등의 컨텍스트를 관리합니다. 먼저 연결 정보를 받아 세션을 제공하는 객체를 src/infra/repo/sa.py 경로에 구성합니다. 구체적인 구현체이므로 인프라 계층에 구현합니다.

from sqlalchemy import create_engine
from sqlalchemy.orm import Session, scoped_session, sessionmaker


class SA:
    def __init__(self, url: str, options: dict) -> None:
        self.engine = create_engine(url, **options)
        self.session_factory = sessionmaker(bind=self.engine, expire_on_commit=False)
        self.scoped_session_factory = scoped_session(self.session_factory)

    @property
    def session(self) -> Session:
        return self.scoped_session_factory()

    def remove_session(self) -> None:
        self.scoped_session_factory.remove()

SQLAlchemy의 사용법에 대한 자세한 설명은 생략하겠습니다. 그리고 SA.session을 사용해서 컨텍스트를 관리하는 객체를 src/infra/repo/sa_context.py 경로에 구현해 줍니다.

import contextlib
from typing import Iterator

from src.domain.repo.i_db_context import IDBContext
from src.infra.repo.sa import SA


class SAContext(IDBContext):
    def __init__(self, sa: SA) -> None:
        self.sa = sa

    @contextlib.contextmanager
    def begin_tx(self) -> Iterator[None]:
        if self.sa.session.in_transaction():
            yield
        else:
            self.sa.session.begin()

            try:
                yield
                self.sa.session.commit()
            except:
                self.sa.session.rollback()
                raise
            finally:
                self.sa.remove_session()

SAContext는 컨텍스트 관리를 SA 타입 객체에 의존하고 있습니다. 그런데 SA.session 객체는 나중에 만들 레파지토리와도 공통으로 사용해야 합니다. 그래야 컨텍스트와 레파지토리에서 같은 세션을 참조할 수 있기 때문입니다. 하지만 그렇다고 begin_tx 메소드를 호출할 때마다 SA.session 객체를 따로 파라메터로 받는 것은 번거롭고 무의미합니다. 그대신 여기서는 생성자로 의존성을 취하는 방법을 선택했습니다.

여기까지 작성한 내용을 커밋하겠습니다.

레파지토리

오래 돌아왔습니다. 드디어 진짜 레파지토리 객체를 만들어 봅시다. 레파지토리 역시, 컨텍스트와 정확히 동일한 이유로 먼저 도메인 계층 src/domain/repo/i_repo.py 경로에 인터페이스부터 선언합니다.

from __future__ import annotations

import abc
import uuid
from typing import Generic, TypeVar

Entity = TypeVar("Entity")


class IRepo(abc.ABC, Generic[Entity]):
    @abc.abstractmethod
    def get(self, id: uuid.UUID) -> Entity | None:
        ...

    @abc.abstractmethod
    def save(self, entity: Entity) -> None:
        ...

    def get_or_error(self, id: uuid.UUID) -> Entity:
        entity = self.get(id)
        if entity is None:
            raise RepositoryNotFoundError(f"{str(id)}")

        return entity

    def is_exist(self, id: uuid.UUID) -> bool:
        return self.get(id) is not None


class RepositoryError(Exception):
    pass


class RepositoryNotFoundError(RepositoryError):
    pass

지금 우리에게 당장 필요한 것은 save 메소드입니다만, 하는김에 몇가지 자주 쓰는 편의용 메소드를 같이 구현해 보았습니다.

캘린더 레파지토리

하나의 레파지토리는 하나의 엔티티(혹은 집계)에 대응해야 합니다. 따라서 IRepo 인터페이스를 확장하는 Calendar 모델 전용 레파지토리의 인터페이스를 src/domain/repo/i_calendar_repo.py 경로에 구현합니다. 이것은 IRepo 인터페이스를 직접 상속하는 구현체는 존재하지 않을 것이라는 말이 됩니다.

from src.domain.entity.calendar import Calendar
from src.domain.repo.i_repo import IRepo


class ICalendarRepo(IRepo[Calendar]):
    ...

이 인터페이스를 가지고 src/infra/repo/sa_calendar_repo.py 경로에 SQLAlchemy 구현체를 만들어 봅시다.

import uuid

from src.domain.entity.calendar import Calendar
from src.domain.repo.i_calendar_repo import ICalendarRepo
from src.infra.repo.sa import SA


class SACalendarRepo(ICalendarRepo):
    def __init__(self, sa: SA) -> None:
        self.sa = sa

    def get(self, id: uuid.UUID) -> Calendar | None:
        return self.sa.session.get(Calendar, id)

    def save(self, entity: Calendar) -> None:
        self.sa.session.add(entity)

컨텍스트와 마찬가지로 레파지토리에서도 SA 객체에 대한 의존성을 생성자를 통해서 취해줍니다. 그리고 SQLAlchemy의 세션 객체를 통해 데이터베이스에서 모델을 조회하거나 저장할 수 있는 코드를 작성합니다. 이제 SACalendarRepo.save(calendar) 메소드를 사용해서 우리의 캘린더 모델을 저장할 수 있게 되었습니다!

여기까지 작성한 내용을 커밋하고, 다음 글에서는 이 모든 것이 통합된 '캘린더 생성 기능'을 정말로 완성해 보겠습니다.