base-repository

English | 한국어

How To Use (사용자 가이드)

목차

0) 범위

1) 모델 & 스키마 정의

# SQLAlchemy ORM
class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    email: Mapped[str]

# Pydantic 스키마
class UserSchema(BaseModel):
    id: int
    name: str
    email: str
    model_config = ConfigDict(from_attributes=True)

2) Repository 구현

@dataclass
class UserFilter(BaseRepoFilter):
    id: int | None = None
    name: str | None = None

class UserRepo(BaseRepository[User, UserSchema]): # (기본) schema(UserSchema) 반환 활성화
    filter_class = UserFilter

옵션

3) 빠른 시작 (Step-by-step)

3.1 세션 바인딩 (중요)

BaseRepository의 세션 우선순위

  1. SessionProvider가 설정되어 있으면, repo는 항상 provider.get_session()을 사용합니다.
  2. Provider가 없으면, repo 생성자에서 전달한 session(specific session)을 사용합니다.
  3. 둘 다 없으면 런타임 에러가 납니다.

추가로, 각 메서드 호출에서 session=...을 전달하면 그 호출에서는 항상 그 세션이 최우선입니다.

3.1.1 (권장) Provider 기반 구성

# 앱 초기화 시점에 1회 설정
UserRepo.configure_session_provider(session_provider)

# 이후 repo는 세션을 직접 들고 있을 필요가 없음
repo = UserRepo()

rows = await repo.get_list(
    flt=UserFilter(name="A"),
    page=1,
    size=10,
)

3.1.2 (옵션) 호출 단위로 세션 직접 주입

repo = UserRepo()

async with AsyncSession(engine) as session:
    rows = await repo.get_list(
        flt=UserFilter(name="A"),
        page=1,
        size=10,
        session=session,  # 이 호출에서만 사용됨
    )

3.1.3 (옵션) Provider 없이 repo에 세션 바인딩

async with AsyncSession(engine) as session:
    repo = UserRepo(session=session)
    rows = await repo.get_list(flt=UserFilter(name="A"), page=1, size=10)

주의

3.2 트랜잭션/커밋 책임 (중요)

3.3 생성(단건/다건)

# 단건: 스키마 또는 dict
created = await repo.create(UserSchema(name="Alice", email="a@test.com"))
created2 = await repo.create({"name": "Bob", "email": "b@test.com"})

# 다건
bulk = await repo.create_many([
    {"name": "C", "email": "c@test.com"},
    UserSchema(name="D", email="d@test.com"),
])

# ORM 모델 직접
created3 = await repo.create_from_model(User(name="E", email="e@test.com"))

PK 처리 규칙

3.4 조회(단건)

# 단건
row = await repo.get(UserFilter(name="Alice"))  # 기본: 도메인 반환 (mapping_schema 설정 시)
row_raw = await repo.get(UserFilter(name="Alice"), convert_schema=False)  # ORM 반환

3.5 조회(다건 - ListQuery 체이닝)

OFFSET 페이징

q = (
    repo.list(flt=UserFilter())
        .order_by([User.id.asc()])   # 선택(미지정 시 구현체에서 보강될 수 있음)
        .paging(page=1, size=20)     # page>=1, size>=1
)

# 또는 where 체이닝
q = (
    repo.list()
        .where(UserFilter())
        .order_by([User.id.asc()])
        .paging(page=1, size=20)
)

rows = await repo.execute(q)         # SELECT는 항상 list 반환

CURSOR(Keyset) 페이징

# 첫 페이지: None 또는 {}
q1 = (
    repo.list()
        .order_by([User.id.asc()])   # 커서 이전에 지정
        .with_cursor(None)
        .limit(20)                   # size>=1
)
rows1 = await repo.execute(q1)

# 다음 페이지
next_cursor = {"id": 123}
q2 = (
    repo.list()
        .order_by([User.id.asc()])
        .with_cursor(next_cursor)    # order_by 컬럼 키/순서와 동일해야 함
        .limit(20)
)
rows2 = await repo.execute(q2)

where 사용 패턴

# 여러 필드 조합
q = repo.list().where(UserFilter(id=1, name="Kyu"))

# list(flt=...)와 .where(...)를 같이 쓰면 where()에서 예외가 날 수 있습니다.
# (list()에서 이미 filter가 설정된 상태이기 때문)

order_by 입력 예시(지원되는 다양한 형태)

# 1) 문자열 키
q = repo.list().order_by(["id"]).paging(page=1, size=10)

# 2) Enum.value (문자열) — 예시 Enum
class SortKey(Enum):
    ID = "id"
    NAME = "name"
q = repo.list().order_by([SortKey.ID, SortKey.NAME]).paging(page=1, size=10)

# 3) 모델 컬럼 속성
q = repo.list().order_by([User.name]).paging(page=1, size=10)

# 4) asc()/desc()
q = repo.list().order_by([User.name.desc(), User.id.asc()]).paging(page=1, size=10)

# 5) ColumnElement (expression)
q = repo.list().order_by([User.name.expression]).paging(page=1, size=10)

# 6) 기본값: order_by 생략
q = repo.list().paging(page=1, size=10)

커서 dict 예시(다중 컬럼)

q = (
  repo.list()
     .order_by([User.id.asc(), User.name.desc()])
     .with_cursor({
         "id": 120,
         "name": "Z",
     })
     .limit(20)
)
rows = await repo.execute(q)

불허

3.6 조회(다건 - get_list)

OFFSET 페이징

rows = await repo.get_list(
    flt=UserFilter(name="A"),
    order_by=[User.id.asc()],
    page=1,
    size=10,
)

rows = await repo.get_list(
    flt=UserFilter(id=1, name="A"),
)

rows = await repo.get_list(
    order_by=["id", User.name.desc()],
    page=2,
    size=10,
)

rows = await repo.get_list(page=1, size=10)

CURSOR(Keyset) 페이징

rows1 = await repo.get_list(
    order_by=[User.id.asc()],
    cursor={},   # 또는 None (이 경우에도 keyset 모드)
    size=10,
)

rows2 = await repo.get_list(
    order_by=[User.id.asc(), User.name.desc()],
    cursor={"id": 120, "name": "Z"},
    size=10,
)

get_list 조합 규칙

  1. cursor를 전달하면(None 포함) keyset 페이징으로 동작합니다.

  1. page와 size를 같이 지정하면 offset 페이징으로 동작합니다.

  2. 그 외는 페이징이 적용되지 않습니다.

3.7 개수 조회

cnt = await repo.count(UserFilter(name="Alice"))

3.8 update

# 일괄 UPDATE (단일 UPDATE SQL)
updated = await repo.update(UserFilter(name="Bob"), {"email": "bob@new"})

# 영속 객체 Dirty Checking
obj = await repo.get(UserFilter(name="Alice"), convert_schema=False)
updated_schema = await repo.update_from_model(obj, {"email": "alice@new"})

3.9 삭제

deleted = await repo.delete(UserFilter(name="Alice"))

4) 공개 API 레퍼런스

4.1 Repository 인스턴스 메서드

4.2 ListQuery 체이닝 메서드

5) BaseRepoFilter

규칙

예시

@dataclass
class UserFilter(BaseRepoFilter):
    id: int | list[int] | None = None
    active: bool | None = None
    __aliases__ = {"org_ids": "org_id"}

6) 매핑(도메인 변환) 옵션

예시

class UserMapper(BaseMapper):
    def to_schema(self, db: User) -> UserSchema:
        return UserSchema(id=db.id, name=db.name, email="Changed")

    def to_orm(self, dm: UserSchema) -> User:
        return User(**dm.model_dump(exclude=["name"]), name="fixed")

class UserRepo(BaseRepository[User, UserSchema]):
    filter_class = UserFilter
    mapper = UserMapper

row = await repo.get(UserFilter(id=1))
assert row.email == "Changed"

6.1 mapper 사용 시 스키마 검증 동작

mapper(BaseMapper)를 지정한 경우, BaseRepository는 mapping_schema 필드 이름과 ORM 모델 컬럼 이름의 1:1 일치 여부를 검증하지 않습니다.

의도

동작 요약

class UserRepo(BaseRepository[User, UserSchema]):
    filter_class = UserFilter
    mapper = UserMapper  # mapper가 존재하면 column-only strict 검증 비활성화

7) 퍼포먼스 테스트

명령어 모음

세팅 방법

  1. CPU
  1. DB
This site is open source. Improve this page.