제네릭 BaseRepository로 쿼리를 가둬 고칠 곳을 한 군데로 모으는 법

데이터 접근은 리포지토리로만: 제네릭 BaseRepository

쿼리를 리포지토리에 가두면 고칠 곳이 한 군데로 준다

서비스 코드를 짜다 보면 손이 먼저 나간다. 목록 하나 뽑고 싶으면 select(...) 한 줄, 단건 하나 가져오려면 session.get(...) 한 줄. 빠르고 편하다. 그런데 이게 도메인마다, 함수마다 흩뿌려지기 시작하면 어느 순간 같은 모양의 쿼리를 열 군데에서 복붙하고 있다. 정렬 규칙 하나 바꾸려면 열 군데를 다 찾아 고쳐야 한다.

그래서 규칙을 하나 세웠다. 데이터에는 리포지토리를 통해서만 접근한다. 서비스와 태스크 본문에는 select, session.execute, session.get, update 같은 원시 쿼리를 쓰지 않는다. 데이터 접근은 전부 리포지토리에 위임하고, 서비스는 “무엇을 가져올지”만 말한다.

같은 CRUD를 매번 다시 짜지 않으려고

문제는, 도메인마다 리포지토리를 만들면 결국 도메인마다 똑같은 CRUD를 또 쓴다는 거다. 조회·필터·정렬·페이지네이션은 어느 엔티티든 모양이 거의 같은데.

그래서 제네릭 기반 클래스 하나로 묶었다. BaseRepository[ModelType] 형태로 선언하면 어떤 모델에도 붙고, 내부에서 QueryBuilder가 필터·정렬·페이지네이션·관계 로딩을 동적으로 조립한다. 도메인 리포지토리는 모델만 지정하면 기본기를 통째로 물려받는다.

class HeroRepository(BaseRepository[Hero]):
    model: Type[Hero] = Hero
    # 기본 CRUD는 한 줄도 안 짜도 다 따라온다

조회는 find_all에 딕셔너리만 넘기면 된다. 필터 키는 "컬럼명:연산자" 형식이고, 연산자를 생략하면 eq다. 지원하는 연산자는 eq/gte/lte/gt/lt/like/startswith/endswith/in/notin.

heroes = await hero_repository.find_all(
    filters={
        "is_active": True,            # 연산자 생략 = eq
        "age:gte": 30,                # 30 이상
        "name:like": "man",          # 이름에 'man' 포함
        "team.name:eq": "정의의 팀",   # To-One 관계는 dot으로
    },
    sorts={"name": "asc"},
    load_with=["powers", "team"],     # N+1 막는 eager loading
)

페이지네이션까지 필요하면 query를 쓴다. 이쪽은 페이지 잘라낸 items와 전체 개수 total_count를 같이 돌려줘서, 목록 API 응답을 만들 때 필요한 게 다 들어 있다.

get이냐 find냐, 없을 때 어떻게 할 건지

같은 단건 조회인데 두 갈래로 나눴다. get(id)는 없으면 NotFoundException을 던지고, find(id)는 없으면 None을 돌려준다. “당연히 있어야 하는데 없으면 에러” 상황엔 get, “있을 수도 없을 수도”엔 find. 호출하는 쪽 코드가 분기를 따로 안 짜도 의도가 드러난다.

존재 확인도 함정이 하나 있다. “있나 없나”를 find_by로 보면 안 된다. find_by는 단일 행만 보장해서 결과가 2건 이상이면 MultipleResultsFound가 터진다. 행이 있는지만 볼 땐 get_total(filters={...}) > 0을 쓴다.

기본기로 안 되는 쿼리는 따로

BaseRepository로 다 되는 건 아니다. IS NOT NULL, 집계(count/sum), bulk UPDATE, JOIN으로 다른 테이블 컬럼만 추출, To-Many 관계 필터 같은 건 표현이 안 된다. 정의되지 않은 연산자를 키에 넣으면 QueryBuilder가 조용히 무시해서 엉뚱한 결과가 나오니 더 위험하다.

이럴 땐 그 모델의 구체 리포지토리에 전용 메서드를 추가하고, 서비스는 그걸 호출만 한다. 기반 클래스는 절대 건드리지 않는다.

async def find_with_ticker(self) -> list[Stocks]:
    """ticker가 NULL 아닌 것만. IS NOT NULL은 기본 필터로 표현 불가."""
    stmt = select(Stocks).where(Stocks.ticker.is_not(None))
    result = await self.session.execute(stmt)
    return list(result.scalars().all())

원시 쿼리를 아예 안 쓰는 게 아니라, 쓰더라도 리포지토리 안에 가둔다. 서비스에서 보면 어디서 가져오든 똑같이 메서드 호출 한 줄이다.

곁가지: 스키마 순환 참조

데이터 모델·스키마를 짜다 보면 서로를 참조하는 순환이 생긴다. 포트폴리오가 보유 종목 목록을 들고, 그 종목이 다시 포트폴리오를 가리키는 식. 이걸 피하는 방식은 세 가지를 조합한다. 스키마 파일 맨 위에 from __future__ import annotations를 두어 타입 힌트를 문자열로 미루고, 다른 파일 스키마는 if TYPE_CHECKING: 블록 안에서만 임포트해 타입 힌트엔 문자열로 적는다. 마지막으로 런타임에 그 문자열을 실제 클래스로 풀어줘야 하니, 모든 Public 스키마의 model_rebuild(force=True, _types_namespace=...) 호출을 app/domain/modules/__init__.py 한 곳에 모아 일괄 처리한다. 모든 모듈이 다 로드된 뒤에 참조를 해석하게 만드는 거다.

핵심만 다시. 데이터 접근은 리포지토리로만, 공통 CRUD는 제네릭 기반 클래스로, 안 되는 쿼리만 구체 리포지토리로 내린다. 서비스가 쿼리를 모르게 하면 나중에 정렬 하나 바꿀 때 고칠 곳이 한 군데로 줄어든다.