Study/fastapi

[udemy] FastAPI - The Complete Course 2025 (Beginner + Advanced) - 학습 정리 1

bluebamus 2025. 1. 6.

강의 : https://www.udemy.com/course/fastapi-the-complete-course/

저장소 : https://github.com/codingwithroby/fastapi-the-complete-course.git

* 기본적으로 알고 있는 내용은 정리를 하지 않고, 추가 정리가 필요한 항목만 정리함

* 강의 내용이 워낙 부실해서 GPT 및 검색 자료를 추가함

1. Project 1 - FastAPI Request Method Logic

- url에서 공백을 표시하는 방법 : %20 은 아스키 코드(ASCII Code)로 변환된 공백 표시이다.

http://localhost:8000/books/title%20one

1) body()는 json으로 전달된 데이터를 파이썬의 딕셔너리 타입으로 변환하여 반환한다.

from fastapi import Body

@app.post("/books/create_book")
async def create_book(new_book=Body()):
    BOOKS.append(new_book)

2) body()를 사용하여 인자를 꺼내는 방법에 대한 코드 예시

@app.put("/books/update_book")
async def update_book(updated_book=Body()):
    for i in range(len(BOOKS)):
        if BOOKS[i].get('title').casefold() == updated_book.get('title').casefold():
            BOOKS[i] = updated_book

2. Project 2 - Move Fast with FastAPI

1) Pydantic v1 vs Pydantic v2 두 버전간의 차이점 정리

- .dict() 함수 이름이 .model_dump()로 변경됨

- schema_extra 함수는 Config class 내에서 json_schema_extra로 변경됨

- Optional variables는 =None가 필요하다 예를 들어 id: Optional[int] = None 와 같이 표현이 가능함

- age: int | None = None 이 더 많이 사용된다.

2) model_config을 사용한 샘플 코드 작성 방법

- v2에서의 사용 방법

model_config = {
        "json_schema_extra": {
            "example": {
                "title": "A new book",
                "author": "codingwithroby",
                "description": "A new description of a book",
                "rating": 5,
                'published_date': 2029
            }
        }
    }

- v1에서의 사용 방법

class Config:
    json_schema_extra = {
        "example": {
            "id": 1,
            "name": "Alice",
            "age": 25
        }
    }

3) from fastapi import Path 에 대한 정리

1. Path의 기본 역할

- 경로 매개변수의 속성을 정의
- 유효성 검사를 위한 제약 조건 설정
- 문서화를 위한 메타데이터 제공(OpenAPI 스키마)

- 기본 사용 예시

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(..., title="The ID of the item")):
    return {"item_id": item_id}

2. Path 매개변수의 속성

- 기본값 정의

@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(42)):
    return {"item_id": item_id}

- 필수 매개변수

- ...을 사용한 필수 매개변수 설정

@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(...)):
    return {"item_id": item_id}

- 설명 및 문서화

- title, description, example 등을 사용해 OpenAPI 문서에 정보 추가

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="Item ID", description="The ID of the item", example=123)
):
    return {"item_id": item_id}

- 값의 범위 제한

- 숫자 매개변수에 대해 최소값, 최대값 등을 제한

- ge=1: item_id는 최소 1이어야 한다.

- le=100: item_id는 최대 100이어야 한다.

- 유효하지 않은 값이 전달되면 422 Unprocessable Entity 에러가 발생한다.

@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(..., ge=1, le=100)):
    return {"item_id": item_id}

- 정규식 제한

- 문자열 매개변수에 대해 정규식을 사용

- regex="^[a-zA-Z0-9_-]{3,16}$": username은 3~16자의 알파벳, 숫자, _, -만 허용

- 유효하지 않은 값이 전달되면 422 Unprocessable Entity 에러가 발생

@app.get("/users/{username}")
async def read_user(username: str = Path(..., regex="^[a-zA-Z0-9_-]{3,16}$")):
    return {"username": username}

3. Path와 Query 매개변수의 조합

- Path는 경로 매개변수에 사용되며, 쿼리 매개변수와 함께 사용할 수 있다.

- 경로 매개변수: item_id는 필수

- 쿼리 매개변수: q는 선택적

from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
async def read_item(
    item_id: int = Path(..., title="The ID of the item", ge=1),
    q: str = None
):
    return {"item_id": item_id, "q": q}

4. Path 매개변수와 데이터 타입

- Path는 Python의 타입 힌트를 기반으로 유효성 검사를 수행

- 문자열

@app.get("/users/{username}")
async def read_user(username: str = Path(...)):
    return {"username": username}

- 정수

@app.get("/items/{item_id}")
async def read_item(item_id: int = Path(..., ge=1)):
    return {"item_id": item_id}

- Enum

from enum import Enum

class ItemType(str, Enum):
    book = "book"
    pen = "pen"

@app.get("/items/{item_type}")
async def read_item(item_type: ItemType = Path(...)):
    return {"item_type": item_type}

5. Query의 기본 사용법

- Query()는 쿼리 매개변수의 기본값을 설정하거나 매개변수를 선택적 또는 필수로 설정할 수 있다.

- q: str: 쿼리 매개변수 q의 데이터 타입은 문자열.
- Query(...): q는 필수 매개변수.
- title: 매개변수의 제목(OpenAPI 문서에 표시됨).
- description: 매개변수의 상세 설명.

from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items/")
async def read_items(q: str = Query(..., title="Query String", description="Search query")):
    return {"q": q}

- 요청 예시

GET /items/?q=example

------------------

{
    "q": "example"
}

6. Query의 주요 옵션

- 쿼리 매개변수에 기본값 설정

@app.get("/items/")
async def read_items(q: str = Query("default value")):
    return {"q": q}

- 필수 매개변수

- ...를 사용해 필수 매개변수 설정

- 쿼리 매개변수가 누락되면 422 Unprocessable Entity 에러 발생

@app.get("/items/")
async def read_items(q: str = Query(...)):
    return {"q": q}

- 제약 조건 설정

- 쿼리 매개변수 값에 대해 길이 또는 범위 제한 가능

- 문자열 길이 제한

- min_length=3: 최소 길이는 3
- max_length=50: 최대 길이는 50
- regex: 알파벳과 숫자만 허용

@app.get("/items/")
async def read_items(
    q: str = Query(..., min_length=3, max_length=50, regex="^[a-zA-Z0-9]+$")
):
    return {"q": q}

- 요청 : GET /items/?q=ab

- 응답 : 422 Unprocessable Entity

GET /items/?q=ab

----------------------

{
    "detail": [
        {
            "loc": ["query", "q"],
            "msg": "ensure this value has at least 3 characters",
            "type": "value_error.any_str.min_length",
        }
    ]
}

- 숫자 범위 제한

- ge=1: 최소값은 1.
- le=100: 최대값은 100.

@app.get("/items/")
async def read_items(
    size: int = Query(..., ge=1, le=100)
):
    return {"size": size}

- 요청: GET /items/?size=150
- 응답: 422 Unprocessable Entity

- 목록 값 처리

- 쿼리 매개변수로 목록 데이터를 받을 수 있음

@app.get("/items/")
async def read_items(tags: list[str] = Query([])):
    return {"tags": tags}

- 요청: GET /items/?tags=tag1&tags=tag2

GET /items/?tags=tag1&tags=tag2

-----------------------

{
    "tags": ["tag1", "tag2"]
}

- 문서화

- Query()를 사용해 OpenAPI 문서에 메타데이터 추가 가능

- title: 매개변수의 제목
- description: 매개변수에 대한 설명
- example: OpenAPI 문서에서 예제로 표시될 값

@app.get("/items/")
async def read_items(
    q: str = Query(..., title="Search Query", description="The search query for items", example="example")
):
    return {"q": q}

7. Query와 선택적 매개변수

- 쿼리 매개변수를 선택적으로 설정하려면 기본값을 None으로 지정

@app.get("/items/")
async def read_items(q: str | None = Query(None)):
    return {"q": q}

8. Annotated 내부에서 Query()를 사용하는 방법

@app.put("/items_json/{item_id}")
async def update_item_json(
    item_id: int = Path(..., gt=0),
    q1: str = Query(None, max_length=50),
    #q1: Annotated[str, Query(max_length=50)] = None
    q2: str = Query(None, min_length=3),
    #q2: Annotated[str, Query(min_length=50)] = None
    item: Item = None
):
    return {"item_id": item_id, "q1": q1, "q2": q2, "item": item}

3. Setup Database

1) Database Connection 설정

1. create_engine :

- SQLAlchemy에서 데이터베이스 연결을 생성하는 함수이다. 데이터베이스 URL을 기반으로 연결을 설정하며, 다양한 옵션을 통해 데이터베이스 연결 동작을 제어할 수 있다.

- 주요 옵션 :

- echo : SQLAlchemy가 실행하는 SQL 쿼리를 출력한다. 디버깅 시 사용하고, 상용 환경에서는 비활성화해야 한다.
- pool_size : 데이터베이스 연결 풀의 기본 크기이다. 기본값은 5이며, 서비스의 부하를 고려하여 적절히 설정한다.
- max_overflow : 풀의 기본 크기를 초과하여 생성할 수 있는 연결 수이다. 기본값은 10이다.
- pool_timeout : 풀에서 연결을 가져오기 위해 기다리는 시간(초)이다.
- pool_recycle : 연결을 재활용하기 전에 유지되는 시간(초)이다. 데이터베이스 연결이 오래 유지될 경우 문제가 발생할 수 있으므로 설정한다.
- future: SQLAlchemy 2.0 스타일 사용을 위한 옵션이다. 최신 버전에서는 반드시 활성화해야 한다.
- connect_args: 데이터베이스 연결에 전달할 추가 인수이다. 예를 들어, SQLite에서는 외부 키 제약 조건을 활성화하기 위해 사용된다.

2. sessionmaker :

- SQLAlchemy 세션을 생성하는 팩토리 함수이다. 세션은 데이터베이스와 상호작용을 관리하며, ORM 작업의 핵심 역할을 한다.

- 주요 옵션 :

- class_ : 사용할 세션 클래스이다. 비동기 환경에서는 AsyncSession을 사용해야 한다.
- bind : 세션이 사용할 데이터베이스 엔진이다.
- expire_on_commit : 세션이 커밋 후 객체를 만료할지 여부를 결정한다. 상용 서비스에서는 False로 설정하여 성능을 향상시킨다.

3. declarative_base :

- SQLAlchemy ORM에서 사용할 모델 클래스의 기본 클래스를 생성한다. 데이터베이스 테이블과 Python 클래스를 매핑하는 데 사용된다.

4. 상용 서비스에 적합한 설정 요약:

- 데이터베이스 풀링 : pool_size와 max_overflow를 설정하여 동시 요청 처리 능력을 향상
- 세션 관리: expire_on_commit=False로 설정하여 세션 커밋 후 객체를 만료하지 않음
- 비동기 엔진 사용: create_async_engine과 AsyncSession을 사용하여 비동기 작업을 처리
- 에코 비활성화: echo=False로 SQL 출력 비활성화
- 커넥션 재활용: pool_recycle로 장시간 유지된 연결의 문제 방지

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
from sqlalchemy.pool import NullPool

# FastAPI 앱 생성
app = FastAPI()

# 데이터베이스 URL (PostgreSQL 예시)
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

# 1. 데이터베이스 엔진 생성
# 상용 서비스에서 주로 사용하는 옵션들을 포함
engine = create_async_engine(
    DATABASE_URL,
    echo=False,  # 디버깅 시 True로 설정하여 SQL 쿼리를 확인 가능
    pool_size=10,  # 기본 연결 풀 크기
    max_overflow=20,  # 최대 추가 연결 수
    pool_timeout=30,  # 연결 풀에서 대기 시간 (초)
    pool_recycle=1800,  # 연결 재활용 시간 (초)
    future=True,  # SQLAlchemy 2.0 스타일 사용
)

# 2. 세션 팩토리 생성
# expire_on_commit=False로 설정하여 커밋 후 객체 만료 방지
AsyncSessionLocal = sessionmaker(
    bind=engine,
    class_=AsyncSession,
    expire_on_commit=False,
)

# 3. Declarative Base 생성
# ORM 모델의 기본 클래스
Base = declarative_base()

# 4. 데이터베이스 세션 의존성
# 요청마다 독립적인 세션을 제공
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
        finally:
            await session.close()

# 5. FastAPI 엔드포인트 예시
@app.get("/")
async def read_root():
    return {"message": "Hello, FastAPI with SQLAlchemy!"}

5. autoflush에 대한 정리

- flush는 databse에 가상적인 변경 사항을 만들고 commit은 실제 변경 사항을 만든다.

- 만약, 동일한 트랜잭션에서 pk와 같은 저장 후의 결과를 사용할 필요가 있다면, add 혹은 update를 실행 후 flush를 이용해 해당 데이터를 가져올 수 있다.

- 하지만, 해당 결과는 database에 반영된 결과가 아니기 때문에 다른 세션 연결에는 보이지 않는다.

- flush를 실행한 이후 rollback을 수행할 수 있지만, commit을 이용해 실재 데이터가 반영된 이후에는 rollback을 수행할 수 없다.

- 비동기 작업에는 autoflush를 False로 설정한다.

- 성능 측면 : 비동기 환경에서는 여러 작업이 동시에 실행되기 때문에, 자동으로 여러 번의 flush()가 발생할 수 있다. 이는 불필요한 I/O 작업을 초래하고 성능 저하를 유발할 수 있다.

- 작업 간 의도치 않은 상호작용: 비동기 작업은 병렬로 실행되기 때문에, 자동으로 세션 상태를 동기화하는 flush()가 예기치 않게 다른 작업에 영향을 미칠 수 있다. 비동기 환경에서는 flush()가 자동으로 발생하는 타이밍에 더 큰 영향을 미칠 수 있기 때문에, 다른 작업과의 상호작용을 예상할 수 없게 만든다.

6. Base를 상속받아 테이블 클래스 만들기

from sqlalchemy import (
    Column,
    Integer,
    String,
    Boolean,
    Float,
    DateTime,
    ForeignKey,
    Text,
    UniqueConstraint,
    Index,
    func,
)
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base

# Base 클래스 생성
Base = declarative_base()

# 사용자 테이블 정의
class User(Base):
    """
    사용자 정보를 저장하는 User 테이블.
    """
    __tablename__ = "users"  # 테이블 이름 설정

    # 메타 설정 (복합 유니크 제약 조건 및 인덱스)
    __table_args__ = (
        UniqueConstraint("username", "email", name="uq_user_username_email"),  # 복합 유니크 제약 조건
        Index("ix_user_email", "email"),  # 이메일 컬럼에 인덱스 추가
    )

    # 기본 키 컬럼
    id = Column(
        Integer,
        primary_key=True,
        autoincrement=True,
        nullable=False,
    )

    # 사용자 이름
    username = Column(
        String(50),
        unique=True,  # 유니크 제약 조건 (단일 필드)
        nullable=False,
        index=True,  # 인덱스 추가
    )

    # 이메일
    email = Column(
        String(255),
        unique=True,  # 유니크 제약 조건 (단일 필드)
        nullable=False,
    )

    # 비밀번호
    password = Column(
        String(255),
        nullable=False,
    )

    # 활성 상태
    is_active = Column(
        Boolean,
        default=True,
        nullable=False,
    )

    # 생성일
    created_at = Column(
        DateTime,
        default=func.now(),
        nullable=False,
    )

    # 업데이트일
    updated_at = Column(
        DateTime,
        default=func.now(),
        onupdate=func.now(),
        nullable=False,
    )

    # 관계 설정 (다대일 관계)
    posts = relationship(
        "Post",
        back_populates="author",
        cascade="all, delete-orphan",
    )


# 게시물 테이블 정의
class Post(Base):
    """
    게시물 정보를 저장하는 Post 테이블.
    """
    __tablename__ = "posts"

    # 메타 설정 (복합 유니크 제약 조건 및 인덱스)
    __table_args__ = (
        UniqueConstraint("title", "author_id", name="uq_post_title_author"),  # 복합 유니크 제약 조건
        Index("ix_post_created_at", "created_at"),  # 생성일에 대한 인덱스
    )

    # 기본 키 컬럼
    id = Column(
        Integer,
        primary_key=True,
        autoincrement=True,
        nullable=False,
    )

    # 제목
    title = Column(
        String(100),
        nullable=False,
    )

    # 내용
    content = Column(
        Text,
        nullable=False,
    )

    # 작성자 ID (외래 키)
    author_id = Column(
        Integer,
        ForeignKey("users.id", ondelete="CASCADE"),
        nullable=False,
    )

    # 작성일
    created_at = Column(
        DateTime,
        default=func.now(),
        nullable=False,
    )

    # 관계 설정 (다대일 관계)
    author = relationship(
        "User",
        back_populates="posts",
    )

7. Base.metadata에 대한 정리

- SQLAlchemy ORM에서 테이블 정의와 매핑 정보를 포함한 메타데이터 객체이다.

- Base.metadata.create_all(bind=engine)은 호출될 때마다 데이터베이스에 정의된 테이블과 ORM에서 정의된 테이블 간의 차이를 확인하고, 존재하지 않는 테이블만 생성한다. 따라서 이미 존재하는 테이블은 다시 생성되지 않는다.

- create_all은 주로 개발 환경에서만 사용하며, 프로덕션 환경에서는 실행되지 않도록 분리해야 한다.

- 테이블 구조 변경(예: 컬럼 추가/삭제)은 반영되지 않으므로, 이를 위해서는 Alembic과 같은 마이그레이션 도구를 사용하는 것이 필요하다.

대상	메서드/속성	설명
Base	metadata	모든 테이블 정의와 매핑 정보를 포함.
Base	metadata.create_all	정의된 모든 테이블을 데이터베이스에 생성.
Base	metadata.drop_all	정의된 모든 테이블을 데이터베이스에서 삭제.
Base	metadata.reflect	데이터베이스에 존재하는 테이블 구조를 반영.
metadata	tables	정의된 모든 테이블 정보를 딕셔너리로 반환.
metadata	sorted_tables	외래 키 관계를 기준으로 정렬된 테이블 리스트를 반환.
metadata	bind	연결된 데이터베이스 엔진.
models	relationship	테이블 간의 관계를 정의.
models	ForeignKey	외래 키를 정의하여 테이블 간의 참조 관계 설정.
models	db_session.add(instance)	ORM 객체를 데이터베이스 세션에 추가.
models	db_session.query(Class)	특정 ORM 클래스에 대해 쿼리 실행.

4. API Request Methods

1) chatgpt가 제안하는 프로젝트 구조

- 일반적으로 FastAPI와 SQLAlchemy를 사용하는 프로젝트는 다음과 같은 디렉토리 구조를 가진다.

project/
├── app/
│   ├── __init__.py
│   ├── main.py                # FastAPI 앱 진입점
│   ├── database.py            # 데이터베이스 설정 및 연결
│   ├── models/
│   │   ├── __init__.py        # Base 및 모델 임포트
│   │   ├── user.py            # User 모델 정의
│   │   ├── post.py            # Post 모델 정의
│   ├── crud/
│   │   ├── __init__.py
│   │   ├── user.py            # User CRUD 함수
│   │   ├── post.py            # Post CRUD 함수
│   ├── schemas/
│   │   ├── __init__.py        # Pydantic 스키마 임포트
│   │   ├── user.py            # User Pydantic 스키마
│   │   ├── post.py            # Post Pydantic 스키마
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py          # 설정 (환경 변수, 설정값 등)
│   │   ├── db_utils.py        # 데이터베이스 초기화 스크립트
│   ├── utils/
│   │   ├── __init__.py
│   │   ├── common.py          # 공통 유틸리티 함수
│   ├── migrations/            # Alembic 마이그레이션 관리
│   ├── scripts/               # 스크립트 (초기화, 시드 데이터 등)
│       ├── init_db.py         # 데이터베이스 초기화 스크립트
│   ├── tests/                 # 테스트 코드
│       ├── test_user.py       # User 테스트 코드
│       ├── test_post.py       # Post 테스트 코드
├── requirements.txt           # 의존성 패키지
├── .env                       # 환경 변수 파일

2) database 생성 및 depends()에 사용될 db 관리 함수 생성과 데이터 가져오기

- databse.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

SQLALCHEMY_DATABASE_URL = 'sqlite:///./todosapp.db'

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={'check_same_thread': False})

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

- main.py

from fastapi import FastAPI, Depends
from typing import Annotated
from sqlalchemy.orm import Session
from models import Base, Todos
from database import engine, SessionLocal
from routers import auth, todos, admin, users

app = FastAPI()

models.Base.metadata.create_all(bind=engine)

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
        
@app.get("/")
async def read_all(db: Annotated[Session, Depends(get_db)]):
    return db.query(Todos).all()

- models.py

from database import Base
from sqlalchemy import Column, Integer, String, Boolean, ForeignKey

class Todos(Base):
    __tablename__ = 'todos'

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String)
    description = Column(String)
    priority = Column(Integer)
    complete = Column(Boolean, default=False)
    owner_id = Column(Integer, ForeignKey("users.id"))

3) depends()의 사용 방법 - 구버전 vs 현재 버전

- 구버전

from sqlalchemy.orm import Session
from fastapi import Depends
from typing import Annotated
...

db_dependency = Annotated[Session, Depends(get_db)]

async def get_user(user: user_dependency, db: db_dependency):
	...

- 현재 버전

from sqlalchemy import Connection
from fastapi import Depends
...

async def get_all_blogs(request: Request, conn: Connection = Depends(context_get_conn)):
	...

5. SQLAlchemy ORM에서 제공하는 query 메서드의 주요 기능과 사용 방법에 대한 정리

1) 주요 기본 쿼리 메서드

1. all()

- 쿼리 결과를 리스트로 반환합니다.
- 데이터가 없으면 빈 리스트를 반환합니다.

2. filter()

- 특정 조건을 만족하는 데이터를 필터링합니다.
- 조건은 SQLAlchemy의 연산자(==, <, >, in_ 등)를 사용합니다.

from sqlalchemy.orm import Session
from sqlalchemy import and_

# 복잡한 조건 쿼리
result = session.query(User).filter(User.age > 20, User.city == "Seoul").all()

# 또는 논리 연산자를 사용하여 표현
result = session.query(User).filter(and_(User.age > 20, User.city == "Seoul")).all()

# filter는 내부적으로 여러 조건을 처리할 때 and_로 결합되기 때문에 
  두 코드는 동일한 SQL 쿼리를 생성

- OR : OR 조건은 sqlalchemy.or_ 함수를 사용하여 여러 조건 중 하나라도 만족하는 결과를 가져온다.

from sqlalchemy.orm import Session
from sqlalchemy import or_

# OR 조건: age가 20보다 크거나 city가 "Seoul"인 사용자
result = session.query(User).filter(
    or_(User.age > 20, User.city == "Seoul")
).all()

# 생성되는 SQL:
# SELECT * FROM user WHERE age > 20 OR city = 'Seoul';

- NOT : NOT 조건은 ~ 연산자를 사용하거나 sqlalchemy.not_ 함수를 사용한다.

from sqlalchemy.orm import Session
from sqlalchemy import not_

# NOT 조건: city가 "Seoul"이 아닌 사용자
result = session.query(User).filter(
    ~User.city == "Seoul"  # NOT User.city = 'Seoul'
).all()

# 또는 not_() 사용
result = session.query(User).filter(
    not_(User.city == "Seoul")
).all()

# 생성되는 SQL:
# SELECT * FROM user WHERE NOT city = 'Seoul';

- IN : IN 조건은 특정 값들 중 하나를 만족하는 경우를 처리할 때 사용한다.

# IN 조건: city가 "Seoul", "Busan", "Daegu" 중 하나인 사용자
result = session.query(User).filter(
    User.city.in_(["Seoul", "Busan", "Daegu"])
).all()

# 생성되는 SQL:
# SELECT * FROM user WHERE city IN ('Seoul', 'Busan', 'Daegu');

3. filter_by()

- 간단한 조건 필터링에 사용됩니다.
- filter와 달리 모델 속성 이름을 직접 키워드로 전달합니다.

3-1. filter_by와 filter의 주요 차이점 요약

특징	filter	filter_by
조건의 복잡성	복잡한 조건 가능 (>, <, !=, and_, 등)	단순 조건만 가능 (=)
표현 방식	SQL 표현식 사용 (User.age > 20)	키워드 인수 사용 (city="Seoul")
사용 목적	유연하고 복잡한 쿼리	간단하고 명확한 쿼리
동작 방식	일반 메서드	키워드 인수를 기반으로 동작

4. first()

- 쿼리 결과의 첫 번째 행을 반환합니다.
- 결과가 없으면 None을 반환합니다.

5. one()

- 정확히 하나의 결과가 반환되는 경우 사용합니다.
- 결과가 없거나 여러 개인 경우 예외를 발생시킵니다.

6. one_or_none()

- 하나의 결과가 있으면 반환하고, 없으면 None을 반환합니다.
- 결과가 여러 개인 경우 예외를 발생시킵니다.

6-1. one()과 one_or_none()의 주요 차이점 요약

특징	one()	one_or_none()
결과가 없을 때	NoResultFound 예외 발생	None 반환
결과가 1개일 때	결과 객체 반환	결과 객체 반환
결과가 2개 이상일 때	MultipleResultsFound 예외 발생	MultipleResultsFound 예외 발생
사용 목적	정확히 하나의 결과를 보장해야 할 때 사용	결과가 없거나 하나일 수 있는 경우 사용

7. count()

- 쿼리 결과의 개수를 반환합니다.

8. order_by()

- 결과를 정렬합니다.
- 정렬 기준은 컬럼 또는 모델 속성을 기준으로 지정합니다.

latest_users = db.query(User).order_by(User.created_at.desc()).limit(10).all()

9. limit()

- 반환할 결과의 최대 개수를 제한합니다.

10. offset()

- 건너뛰기: 쿼리 결과에서 지정한 숫자만큼의 행을 건너뜁니다.

- 0 기반 인덱스: offset(0)은 결과를 건너뛰지 않으며, offset(n)은 처음 n개의 행을 건너뜁니다.

- 페이징: 데이터의 일부분만 조회할 때, limit()과 함께 사용하여 특정 페이지의 데이터를 가져오는 데 활용됩니다.

- 예제 1: 단독 사용

# offset()만 사용하는 경우
result = session.query(User).order_by(User.id).offset(2).all()

# SQL 쿼리:
# SELECT * FROM user ORDER BY id OFFSET 2;

- 예제 2: offset() + limit()로 페이징 구현

- 페이지 크기(page_size): 5
- 페이지 번호(page_number): 2

# 페이지 크기와 페이지 번호
page_size = 5
page_number = 2  # 2번째 페이지

# offset()과 limit()을 사용한 페이징
result = session.query(User).order_by(User.id).limit(page_size).offset((page_number - 1) * page_size).all()

# SQL 쿼리:
# SELECT * FROM user ORDER BY id LIMIT 5 OFFSET 5;

- 예제 3: offset()과 조건 결합

- age > 20 조건을 만족하는 사용자 중 이름 순으로 정렬하고, 처음 3개의 결과를 건너뜁니다.

result = session.query(User).filter(User.age > 20).order_by(User.name).offset(3).all()

# SQL 쿼리:
# SELECT * FROM user WHERE age > 20 ORDER BY name OFFSET 3;

- 예제 4: 페이징 로직 구현

def get_paginated_results(query, page_size, page_number):
    return query.limit(page_size).offset((page_number - 1) * page_size).all()

page_size = 10
page_number = 3
query = session.query(User).order_by(User.id)

results = get_paginated_results(query, page_size, page_number)

# SQL 쿼리:
# SELECT * FROM user ORDER BY id LIMIT 10 OFFSET 20;

- 예제 4: 페이징 로직 구현

- 이 방식은 OFFSET을 사용하지 않아도 되며, 성능이 더 좋습니다.

last_seen_id = 100
page_size = 10

result = session.query(User).filter(User.id > last_seen_id).order_by(User.id).limit(page_size).all()

# SQL 쿼리:
# SELECT * FROM user WHERE id > 100 ORDER BY id LIMIT 10;

11. join()

- 다른 테이블과 조인을 수행합니다.

user_posts = db.query(User, Post).join(Post, User.id == Post.user_id).filter(Post.is_published == True).all()

12. group_by()

- 데이터를 그룹화합니다.

age_groups = db.query(User.age, func.count(User.id)).group_by(User.age).all()

13. distinct()

- 중복을 제거한 고유 값을 반환합니다.

14. scalar()

- 쿼리의 결과에서 단일 값(스칼라 값)을 반환합니다.

- 쿼리의 결과를 리스트나 튜플 대신 단일 값(상수, 문자열 등)으로 변환하여 반환한다.

- 주로 집계 함수(예: COUNT, SUM, AVG)나 단일 열 값을 반환하는 쿼리에서 사용된다.

- query.all()은 결과를 리스트로 반환.
- query.scalar()는 결과를 단일 값으로 반환.

1. 쿼리 결과가 리스트인 경우 (all())

result = session.query(User.name).filter(User.id == 1).all()
# 출력: [("Alice",)]

2. 쿼리 결과가 튜플인 경우 (first())

result = session.query(User.name).filter(User.id == 1).first()
# 출력: ("Alice",)

3. 단일 값만 필요할 때 (scalar())

result = session.query(User.name).filter(User.id == 1).scalar()
# 출력: "Alice"

15. 기본 쿼리 사용 예제

from sqlalchemy.orm import Session
from sqlalchemy import func, or_
from my_models import User, Post  # 예시 ORM 모델

async def get_user(user: User, db: Session):
    """
    사용자 데이터를 쿼리하여 다양한 작업을 수행하는 예제 함수.
    
    :param user: 조회 조건에 사용될 User 객체
    :param db: SQLAlchemy Session 객체
    :return: 쿼리 결과를 처리한 리스트
    """
    # 1. 모든 사용자 조회
    all_users = db.query(User).all()
    # 결과: User 객체의 모든 행 리스트

    # 2. 특정 조건으로 필터링
    filtered_users = db.query(User).filter(User.age > 18).all()
    # 결과: age가 18보다 큰 User 객체 리스트

    # 3. 간단한 필터링 (filter_by)
    user_by_name = db.query(User).filter_by(name="John").first()
    # 결과: name이 "John"인 첫 번째 User 객체 (없으면 None 반환)

    # 4. 특정 필드 값만 가져오기
    user_names = db.query(User.name).all()
    # 결과: 모든 사용자의 이름 리스트 (튜플로 반환됨)

    # 5. 정렬된 사용자 조회
    sorted_users = db.query(User).order_by(User.created_at.desc()).limit(5).all()
    # 결과: 생성일(created_at) 기준으로 최신 5명 반환

    # 6. 집계 데이터 (예: 사용자 평균 나이)
    avg_age = db.query(func.avg(User.age)).scalar()
    # 결과: 사용자 나이의 평균값 반환

    # 7. 조인하여 다른 테이블 데이터와 함께 조회
    joined_data = db.query(User, Post).join(Post, User.id == Post.user_id).all()
    # 결과: User와 연결된 Post 데이터를 함께 반환

    # 8. OR 조건 필터링
    or_filtered_users = db.query(User).filter(or_(User.age < 18, User.is_active == True)).all()
    # 결과: 나이가 18 미만이거나 활성 상태인 사용자 반환

    # 9. 중복 제거
    distinct_emails = db.query(User.email).distinct().all()
    # 결과: 중복되지 않은 이메일 리스트

    # 10. 그룹화와 집계
    group_data = db.query(User.age, func.count(User.id)).group_by(User.age).all()
    # 결과: 나이별 사용자 수 반환 (튜플 리스트)

    # 후처리: 데이터 정리 및 가공
    result = [
        {
            "id": user.id,
            "name": user.name,
            "age": user.age,
            "email": user.email,
        }
        for user in filtered_users
    ]

    return result

2) 윈도우 함수(Window Function) 사용

1. sqlalchemy.sql.functions 정리

- sqlalchemy.sql.functions는 SQL 함수(예: SUM, AVG, COUNT, 등)를 표현하는 데 사용된다.

- SQLAlchemy는 내장된 함수와 사용자 정의 함수를 포함하여 다양한 SQL 함수를 지원한다.

- 주요 기능 :

- 내장 SQL 함수 지원: SQLAlchemy는 일반적으로 사용하는 SQL 함수에 대한 객체를 제공한다.

- 예: func.sum, func.avg, func.count 등은 SQL 함수 호출을 표현한다.
- 유연성: SQLAlchemy의 func는 표준 SQL뿐만 아니라 데이터베이스 전용 함수도 지원한다.

- 예: func.now()는 현재 시간을 반환하는 SQL 표현식을 생성한다.
- 사용자 정의 함수: 데이터베이스에 정의된 사용자 함수도 func.<function_name>으로 호출할 수 있다.

from sqlalchemy import func, select
from sqlalchemy.sql import table, column

# 가상 테이블 정의
my_table = table('my_table', column('value'))

# SUM 함수 예제
stmt = select(func.sum(my_table.c.value))
print(stmt)  # SELECT SUM(my_table.value) FROM my_table

2. sqlalchemy.sql.expression.over 정리

- sqlalchemy.sql.expression.over는 **윈도우 함수(Window Function)**를 SQL 표현식으로 작성하는 데 사용된다.

- 윈도우 함수는 SQL에서 특정 범위(윈도우) 내에서 집계 결과를 계산하거나 순위를 매기는 데 사용된다.
- 주요 기능 :
- 윈도우 함수 정의: 집계 함수와 함께 OVER 절을 추가하여 윈도우 함수를 생성한다.
- 파티셔닝 및 정렬 지원: partition_by와 order_by 옵션으로 데이터를 그룹화하거나 정렬된 범위 내에서 계산을 수행할 수 있다.
- SQL 표현식과의 통합: 다른 SQLAlchemy 표현식과 쉽게 결합된다.

from sqlalchemy import func, select
from sqlalchemy.sql.expression import over
from sqlalchemy.sql import table, column

# 가상 테이블 정의
my_table = table('my_table', column('value'), column('category'))

# 윈도우 함수 예제
stmt = select(
    my_table.c.value,
    over(func.sum(my_table.c.value), partition_by=my_table.c.category)
        .label('category_sum')
)
print(stmt)
# 출력: SELECT my_table.value, SUM(my_table.value) OVER (PARTITION BY my_table.category) AS category_sum FROM my_table

3. 주요 윈도우 함수

- 일반적으로 사용되는 함수:
- ROW_NUMBER(): 각 파티션 내의 고유한 행 번호.
- RANK(): 각 파티션 내의 순위 (동점 허용).
- DENSE_RANK(): 동점이 있어도 순위를 건너뛰지 않음.
- SUM(): 파티션 내의 합계.
- AVG(): 파티션 내의 평균.
- COUNT(): 파티션 내의 개수.

- SQLAlchemy에서 지원하는 윈도우 함수
- func.row_number()
- func.rank()
- func.dense_rank()
- func.sum(column)
- func.avg(column)
- func.count(column)

4. sqlalchemy.sql.functions와 sqlalchemy.sql.expression.over의 결합 사용

- 시나리오: 쇼핑몰 데이터
- 쇼핑몰에서는 orders 테이블과 products 테이블이 있습니다.
- 테이블 구조 (예시):
- orders: 고객의 주문 데이터를 포함.
- order_id: 주문 ID
- customer_id: 고객 ID
- order_date: 주문 날짜
- total_amount: 주문 금액

- products: 제품 정보
- product_id: 제품 ID
- category: 제품 카테고리
- price: 제품 가격

- order_items: 각 주문의 제품 정보
- order_id: 주문 ID
- product_id: 제품 ID
- quantity: 제품 수량
- unit_price: 단가

1. 고객별 연간 총 매출 계산
- 목표: 각 고객의 연간 총 매출을 계산하고, 그 중 가장 높은 매출을 기록한 고객을 찾기

from sqlalchemy import func, select, extract
from sqlalchemy.sql.expression import over
from sqlalchemy.orm import aliased
from datetime import date

# 테이블 정의
orders = table('orders', column('order_id'), column('customer_id'), column('order_date'), column('total_amount'))
order_items = table('order_items', column('order_id'), column('product_id'), column('quantity'), column('unit_price'))

# 각 고객의 연간 총 매출 계산 (연도별로 PARTITION BY)
stmt = select(
    orders.c.customer_id,
    func.sum(orders.c.total_amount).over(
        partition_by=[orders.c.customer_id, extract('year', orders.c.order_date)]
    ).label('annual_revenue')
).where(
    extract('year', orders.c.order_date) == 2024
).order_by(orders.c.customer_id)

# 결과 쿼리 출력
print(str(stmt))

SELECT orders.customer_id, SUM(orders.total_amount) OVER (PARTITION BY orders.customer_id, EXTRACT(YEAR FROM orders.order_date)) AS annual_revenue
FROM orders
WHERE EXTRACT(YEAR FROM orders.order_date) = 2024
ORDER BY orders.customer_id;

customer_id	annual_revenue
1	15000
2	20000
3	30000

2. 상품 카테고리별 월별 판매량 및 순위
- 목표: 각 상품 카테고리의 월별 판매량을 계산하고, 판매량에 따라 순위를 매긴다.

stmt = select(
    products.c.category,
    extract('month', orders.c.order_date).label('month'),
    func.sum(order_items.c.quantity).label('total_sales'),
    over(func.sum(order_items.c.quantity), partition_by=[products.c.category, extract('month', orders.c.order_date)], order_by=func.sum(order_items.c.quantity).desc()).label('sales_rank')
).join(order_items, order_items.c.order_id == orders.c.order_id).join(products, products.c.product_id == order_items.c.product_id).group_by(
    products.c.category, extract('month', orders.c.order_date)
).order_by(products.c.category, 'month')

# 결과 쿼리 출력
print(str(stmt))

SELECT products.category, EXTRACT(MONTH FROM orders.order_date) AS month, SUM(order_items.quantity) AS total_sales,
       SUM(order_items.quantity) OVER (PARTITION BY products.category, EXTRACT(MONTH FROM orders.order_date) ORDER BY SUM(order_items.quantity) DESC) AS sales_rank
FROM orders
JOIN order_items ON order_items.order_id = orders.order_id
JOIN products ON products.product_id = order_items.product_id
GROUP BY products.category, EXTRACT(MONTH FROM orders.order_date)
ORDER BY products.category, month;

category	month	total_sales	sales_rank
Electronics	1	5000	1
Electronics	2	4500	2
Clothing	1	6000	1
Clothing	2	4000	2

3. 주문 날짜별 평균 주문 금액
- 목표: 각 주문 날짜별 평균 주문 금액을 계산하고, 전체 평균보다 높은 주문 날짜를 필터링

stmt = select(
    orders.c.order_date,
    func.avg(orders.c.total_amount).label('avg_order_amount'),
    over(func.avg(orders.c.total_amount), partition_by=extract('year', orders.c.order_date)).label('overall_avg')
).group_by(orders.c.order_date).having(
    func.avg(orders.c.total_amount) > over(func.avg(orders.c.total_amount), partition_by=extract('year', orders.c.order_date))
)

# 결과 쿼리 출력
print(str(stmt))

SELECT orders.order_date, AVG(orders.total_amount) AS avg_order_amount,
       AVG(orders.total_amount) OVER (PARTITION BY EXTRACT(YEAR FROM orders.order_date)) AS overall_avg
FROM orders
GROUP BY orders.order_date
HAVING AVG(orders.total_amount) > AVG(orders.total_amount) OVER (PARTITION BY EXTRACT(YEAR FROM orders.order_date));

order_date	avg_order_amount	overall_avg
2024-01-01	150	120
2024-02-01	180	120

4. 고객별 월별 구매 평균 금액 계산 및 순위
- 목표: 각 고객의 월별 평균 구매 금액을 계산하고, 그 중 상위 고객을 추출

stmt = select(
    orders.c.customer_id,
    extract('month', orders.c.order_date).label('month'),
    func.avg(orders.c.total_amount).label('avg_monthly_spent'),
    over(func.avg(orders.c.total_amount), partition_by=orders.c.customer_id, order_by=func.avg(orders.c.total_amount).desc()).label('spending_rank')
).group_by(orders.c.customer_id, extract('month', orders.c.order_date))

# 결과 쿼리 출력
print(str(stmt))

SELECT orders.customer_id, EXTRACT(MONTH FROM orders.order_date) AS month, AVG(orders.total_amount) AS avg_monthly_spent,
       AVG(orders.total_amount) OVER (PARTITION BY orders.customer_id ORDER BY AVG(orders.total_amount) DESC) AS spending_rank
FROM orders
GROUP BY orders.customer_id, EXTRACT(MONTH FROM orders.order_date);

customer_id	month	avg_monthly_spent	spending_rank
1	1	150	1
2	2	200	2
1	2	180	1

5. 연간 총 매출 대비 각 상품의 기여도 계산
- 목표: 각 상품이 연간 매출에서 차지하는 비율을 계산하고, 그 비율에 따라 순위를 매긴다.

stmt = select(
    products.c.product_id,
    func.sum(order_items.c.quantity * order_items.c.unit_price).label('total_sales'),
    over(func.sum(order_items.c.quantity * order_items.c.unit_price), partition_by=extract('year', orders.c.order_date)).label('annual_revenue'),
    (func.sum(order_items.c.quantity * order_items.c.unit_price) / 
     over(func.sum(order_items.c.quantity * order_items.c.unit_price), partition_by=extract('year', orders.c.order_date))
    ).label('contribution_ratio')
).join(order_items, order_items.c.product_id == products.c.product_id).join(orders, orders.c.order_id == order_items.c.order_id).group_by(
    products.c.product_id, extract('year', orders.c.order_date)
)

# 결과 쿼리 출력
print(str(stmt))

SELECT products.product_id, SUM(order_items.quantity * order_items.unit_price) AS total_sales,
       SUM(order_items.quantity * order_items.unit_price) OVER (PARTITION BY EXTRACT(YEAR FROM orders.order_date)) AS annual_revenue,
       (SUM(order_items.quantity * order_items.unit_price) / 
        SUM(order_items.quantity * order_items.unit_price) OVER (PARTITION BY EXTRACT(YEAR FROM orders.order_date))) AS contribution_ratio
FROM products
JOIN order_items ON order_items.product_id = products.product_id
JOIN orders ON orders.order_id = order_items.order_id
GROUP BY products.product_id, EXTRACT(YEAR FROM orders.order_date);

product_id	total_sales	annual_revenue	contribution_ratio
101	50000	200000	0.25
102	70000	200000	0.35
103	30000	200000	0.15

저작자표시 비영리 동일조건

'Study > fastapi' 카테고리의 다른 글

[udemy] FastAPI - The Complete Course 2025 (Beginner + Advanced) - 학습 정리 3 (0)	2025.01.14
[udemy] FastAPI - The Complete Course 2025 (Beginner + Advanced) - 학습 정리 2 (0)	2025.01.10
[인프런] FastAPI 완벽 가이드 - 강의 소감 (1)	2024.12.31
[인프런] FastAPI 완벽 가이드 - 학습 정리 (3) (0)	2024.12.23
[인프런] FastAPI 완벽 가이드 - 학습 정리 (2) (0)	2024.12.18