Tortoise-ORM

김기훈·2025년 10월 24일

DataBase

목록 보기

3/6

Tortoise-ORM은 비동기 ORM(Object-Relational Mapper)이라서,
- Django ORM과 문법이 비슷하지만 await를 붙여서 비동기 쿼리를 수행

예시

# 모든 유저 보기
users = await User.all()

# 활성화된 유저 중 이메일만 보기
emails = await User.filter(is_active=True).values_list("email", flat=True)

# 특정 이름으로 유저 검색
user = await User.get_or_none(username="kihoon")

# 유저 존재 확인
exists = await User.filter(email="test@example.com").exists()

# 유저 이름 수정
await User.filter(id=1).update(username="new_name")

# 유저 삭제
await User.filter(id=2).delete()

특징

항목	설명
비동기(async) 지원	`await`로 DB 쿼리를 실행. FastAPI와 완벽하게 호환.
Django ORM 스타일 문법	`.filter()`, `.get()`, `.all()` 등 직관적인 체인 방식
데이터베이스 독립적	MySQL, PostgreSQL, SQLite 등 여러 DB 지원
Pydantic 연동 쉬움	`.from_queryset()` 으로 Pydantic 모델 자동 변환 가능
자동 마이그레이션 도구	`aerich` 로 스키마 자동 생성 및 버전 관리 가능

구성요소

구조 분리 패턴

app/
├── models/          # ORM 모델 정의
├── schemas/         # Pydantic DTO
├── repositories/    # DB 접근 (UserRepository 등)
├── services/        # 비즈니스 로직
└── api/v1/          # FastAPI 라우터

Model (테이블 정의)

from tortoise import fields
from tortoise.models import Model

class User(Model):
    id = fields.IntField(pk=True)
    username = fields.CharField(max_length=50, unique=True)
    email = fields.CharField(max_length=100, unique=True)
    password = fields.CharField(max_length=255)

    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "user"

    def __str__(self):
        return self.username

class Meta: 와 def str(self)

Tortoise ORM (혹은 Django ORM) 에서 모델의 “부가 설정”과 “출력 형식”을 정의하는 부분

class Meta: — 모델의 메타데이터(설정값) 정의
- “이 모델이 실제 DB에서 어떤 이름으로, 어떤 정렬 기준으로 관리될지 설정하는 부분”

옵션	설명	예시
`table`	실제 DB에 생성될 테이블 이름	`table = "user"` → 테이블명이 `"user"` 로 지정됨
`ordering`	기본 정렬 기준 지정	`ordering = ["-created_at"]` → 최신 순으로 정렬
`table_description`	테이블 설명문 (문서화용)	`"User account table"`
`unique_together`	두 컬럼의 조합이 유일해야 함	`unique_together = ("username", "email")`
`indexes`	인덱스 생성	`indexes = ["username"]`
`abstract`	이 모델이 추상 베이스 클래스임을 지정	`abstract = True` (DB에 테이블 생성 X)

def str(self): — 모델 객체의 문자열 표현

# 1
def __str__(self):
    return self.username

## 결과
user = await User.get(username="kihoon")
print(user)                # 출력: kihoon

# 2 
class Diary(Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=100)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "diary"
        ordering = ["-created_at"]

    def __str__(self):
        return f"Diary<{self.id}>: {self.title}"

## 결과
diary = await Diary.create(title="첫 번째 일기")
print(diary)         # → Diary<1>: 첫 번째 일기

관계 정의 (1:1, 1:N, N:M)

class Post(models.Model):
    id = fields.IntField(pk=True)
    title = fields.CharField(max_length=200)
    author = fields.ForeignKeyField("models.User", related_name="posts")  # N:1

ForeignKeyField → N:1 관계
related_name="posts" → User에서 user.posts 로 접근 가능
반대로 Post에서 post.author 로 접근 가능

DB 초기화 (register_tortoise)

FastAPI에 연결할 때

from tortoise.contrib.fastapi import register_tortoise

register_tortoise(
    app,
    db_url="postgres://user:password@localhost:5432/dbname",
    modules={"models": ["app.models.user", "app.models.post"]},
    generate_schemas=True,
    add_exception_handlers=True,
)

옵션	설명
`db_url`	DB 연결 주소
`modules`	모델 파일 경로 등록
`generate_schemas`	앱 시작 시 자동 테이블 생성
`add_exception_handlers`	예외를 FastAPI 에러로 자동 변환

CRUD 쿼리

user = await User.create(username="kihoon")
user = await User.get_or_none(username="kihoon")
users = await User.filter(is_active=True)
await User.filter(id=1).update(email="new@mail.com")
await User.filter(is_active=False).delete()

관계 조회 (select_related, prefetch_related)

post = await Post.filter(id=1).select_related("author").first()
print(post.author.username)

select_related: N:1, 1:1 관계 (조인)
prefetch_related: 1:N, N:M 관계 (두 번 쿼리)

마이그레이션 (Aerich)

poetry add aerich
aerich init -t app.main.TORTOISE_ORM # 초기화
aerich migrate --name "create_user_table" # 마이그레이션 생성
aerich upgrade # 적용

Pydantic과의 연동

from tortoise.contrib.pydantic import pydantic_model_creator

UserOut = pydantic_model_creator(User, name="UserOut", exclude=("password",))

@app.get("/users")
async def list_users():
    return await UserOut.from_queryset(User.all())

# SQL 쿼리 결과를 Pydantic 모델 형태(JSON 직렬화 가능)로 바로 반환 가능

트랜잭션 처리

from tortoise.transactions import in_transaction

async with in_transaction() as connection:
    await connection.execute_query("DELETE FROM users WHERE id=1;")

메서드

조회(Read)

메서드	설명	예시
`all()`	모든 레코드 조회	`await User.all()`
`filter()`	조건에 맞는 여러 레코드 조회	`await User.filter(is_active=True)`
`exclude()`	특정 조건 제외	`await User.exclude(is_staff=True)`
`get()`	조건에 맞는 하나의 레코드 조회, 없으면 예외 발생	`await User.get(id=1)`
`get_or_none()`	조건에 맞는 하나의 레코드 조회, 없으면 `None` 반환	`await User.get_or_none(username="kihoon")`
`first()`	첫 번째 레코드 반환	`await User.filter(active=True).first()`
`values()`	지정한 컬럼만 dict 형태로 반환	`await User.all().values("id", "username")`
`values_list()`	지정한 컬럼만 튜플 형태로 반환	`await User.all().values_list("id", "email")`
`annotate()`	집계 함수(Count, Avg 등) 추가	`await User.annotate(post_count=Count("posts"))`
`prefetch_related()`	1:N, M:N 관계 미리 불러오기	`await User.all().prefetch_related("posts")`
`select_related()`	1:1, N:1 관계 미리 불러오기	`await Post.all().select_related("author")`
`exists()`	조건에 맞는 레코드 존재 여부 (True/False)	`await User.filter(username="kihoon").exists()`
`count()`	개수 세기	`await User.filter(is_active=True).count()`

생성(Create)

메서드	설명	예시
`create()`	새 레코드 생성	`await User.create(username="kihoon", email="test@example.com")`
`bulk_create()`	여러 개 한 번에 삽입	`await User.bulk_create([User(username="a"), User(username="b")])`

수정(Update)

메서드	설명	예시
`update()`	조건에 맞는 레코드 수정 (QuerySet 방식)	`await User.filter(id=1).update(email="new@mail.com")`
`save()`	객체를 직접 수정 후 저장	`python user = await User.get(id=1); user.email = "new@mail.com"; await user.save()`

삭제(Delete)

메서드	설명	예시
`delete()`	조건에 맞는 레코드 삭제	`await User.filter(is_active=False).delete()`
(객체 메서드)	특정 객체 삭제	`user = await User.get(id=1); await user.delete()`

정렬 / 슬라이싱 / 제한

메서드	설명	예시
`order_by()`	정렬 (오름차순 기본)	`await User.all().order_by("created_at")`
`order_by("-id")`	내림차순	`await User.all().order_by("-id")`
슬라이싱	결과 제한	`await User.all().limit(10)` 또는 `await User.all().offset(5)`

트랜잭션 관련

메서드	설명	예시
`in_transaction()`	트랜잭션 블록 실행	`python async with in_transaction() as conn: await conn.execute_query(...)`
`atomic()`	데코레이터 버전 트랜잭션	`@atomic() async def create_safely(...): ...`

기타 고급 쿼리

메서드	설명	예시
`distinct()`	중복 제거	`await User.all().distinct()`
`only()`	특정 필드만 로드	`await User.all().only("id", "username")`
`using_db()`	특정 DB 연결에서 실행	`await User.using_db("replica").all()`
`raw()`	직접 SQL 실행	`await User.raw("SELECT * FROM users WHERE id = %s", [1])`

FK(외래키, Foreign Key) 관계 설정 방식

Tortoise-ORM에서 외래키(ForeignKey) 는 다른 모델과의 1:N 관계(One-to-Many) 를 의미

ForeignKeyField("models.User")
- 이 필드는 User 테이블의 기본키(id)를 참조하는 외래키(FK)를 가지며,
- Diary 테이블에 user_id 컬럼이 자동으로 생성

옵션

옵션	설명
`related_name`	역참조 시 사용할 이름
`on_delete`	부모 삭제 시 자식 처리 방식 (`CASCADE`, `SET_NULL`, `RESTRICT` 등)
`null=True`	외래키 필드를 비워둘 수 있는지 여부
`default=...`	기본값 설정 가능

다른 관계

관계	예시	설명
1:N	User → Diary	한 유저가 여러 일기를 가짐
N:N	User ↔ Tag	중간 테이블로 `ManyToManyField` 사용
1:1	User ↔ Profile	`OneToOneField` 사용

profile = fields.OneToOneField("models.Profile", related_name="user")
tags = fields.ManyToManyField("models.Tag", related_name="diaries")

김기훈

안녕하세요.

이전 포스트

PostgreSQL

다음 포스트

Tortoise-ORM

DataBase

예시

특징

구성요소

구조 분리 패턴

Model (테이블 정의)

class Meta: 와 def str(self)

관계 정의 (1:1, 1:N, N:M)

DB 초기화 (register_tortoise)

FastAPI에 연결할 때

CRUD 쿼리

관계 조회 (select_related, prefetch_related)

마이그레이션 (Aerich)

Pydantic과의 연동

트랜잭션 처리

메서드

조회(Read)

생성(Create)

수정(Update)

삭제(Delete)

정렬 / 슬라이싱 / 제한

트랜잭션 관련

기타 고급 쿼리

FK(외래키, Foreign Key) 관계 설정 방식

옵션

다른 관계

PostgreSQL

SQL

0개의 댓글