개발 일기

- 레벨업 파이썬 : https://wikidocs.net/book/4170

- 파이썬 강좌 : https://wikidocs.net/book/2251

- 파이썬 라이브러리 탐험 : https://wikidocs.net/book/14021

- 파이썬 틀린코드 : https://wikidocs.net/book/8131

- 파이썬 클린 아키텍처 : https://wikidocs.net/book/9408

스택(Stack)과 힙(Heap)은 프로그래밍에서 중요한 메모리 구조임

이들은 데이터를 저장하고 관리하는 데 사용되며, 주로 컴퓨터 프로그램의 실행 중에 데이터를 저장하는 데 쓰임

여기에서는 이 두 가지 구조의 기본적인 개념과 차이점에 대해 설명할 예정

스택(Stack)

스택은 후입선출(LIFO, Last In, First Out)의 구조를 가진 메모리 영역

이는 마치 접시를 쌓는 것처럼, 마지막에 들어온 데이터가 먼저 처리되는 구조

• 특징
  • 구조: 스택은 선형 자료구조로, 데이터가 일렬로 쌓여있는 형태로, 맨 위에 있는 요소에만 접근 가능
  • 연산: 스택에는 데이터를 스택에 추가하는 "push"와, 스택에서 데이터를 꺼내는 "pop"이라는 두 가지 연산이 존재
  • 메모리 할당방식: 정적 메모리 할당을 사용하여 크기가 컴파일 시에 결정됨
  • 용도: 함수 호출 시에 지역 변수, 매개변수, 반환 주소 등을 저장하는 데 사용되며, 재귀 알고리즘의 실행 흐름을 관리하는 데에도 사용
• 장점
  • 빠른 접근 속도: 메모리에 연속적으로 저장되어 있으므로, 데이터에 빠르게 접근 가능
  • 간편한 관리: 후입선출구조로인해 데이터의 추가와 삭제가 간단하며 메모리 관리에 용이
  • 스택프레임: 함수 호출 시에 지역변수와 매개변수를 저장하는 스택 프레임을 통해 함수의 호출과 반환을 효율적으로 관리 가능
• 단점
  • 크기 제한: 스택의 크기는 컴파일 시에 결정되므로 런타임 시에 동적으로 조정이 불가 → 정적으로 할당된 메모리의 한계로 인해 스택오버플로우가 발생 가능
  • 임시저장만 가능: 스택은 임시데이터를 저장하는 데 사용되므로, 장기적인 데이터 보관에는 적합하지 않음

# 파이썬 리스트를 이용한 스택 구현 예시
class Stack:
    def __init__(self):
        self.stack = []

    def push(self, item):
        self.stack.append(item)

    def pop(self):
        if self.is_empty():
            return None
        return self.stack.pop()

    def peek(self):
        if self.is_empty():
            return None
        return self.stack[-1]

    def is_empty(self):
        return len(self.stack) == 0

# 스택 테스트
stack = Stack()
stack.push(1)
stack.push(2)
stack.push(3)

# 스택은 무조건 가장 나중에 삽입한 값부터 반환함
print("스택의 맨 위 항목:", stack.peek())  # 출력: 3
print("스택에서 팝된 항목:", stack.pop())  # 출력: 3
print("스택의 맨 위 항목:", stack.peek())  # 출력: 2

힙 (Heap)

힙은 메모리의 동적 할당 영역으로 데이터의 저장 및 관리를 위해 사용됨

힙은 데이터를 임의의 순서로 저장하며 메모리가 동적으로 할당되고 해제됨

맨 위의 요소에만 접근 가능한 스택과는 달리, 힙은 임의의 요소에 접근이 가능함

• 특징
  • 구조: 힙은 트리 형태의 자료구조로, 노드들이 부모-자식 관계로 연결되며, 각 노드는 일반적으로 특정한 데이터를 저장
  • 연산: 힙은 데이터를 삽입하는 "insert" 연산과, 삭제하는 "delete" 연산이 존재
  • 메모리 할당방식: 동적 메모리 할당을 사용하여 런타임 중에 크기가 결정됨
  • 용도: 동적으로 크기가 변하는 데이터 구조를 관리하는 데 사용되며, 주로 동적으로 할당된 객체나 배열, 그래프 등을 저장하기 위해 활용
• 장점
  • 동적 할당: 힙은 런타임 중에 메모리를 동적으로 할당하므로, 유연한 데이터 구조를 생성 가능하며 크기제한이 없음
  • 오랜 기간 데이터 보존 가능: 힙은 데이터의 수명이 프로그램의 실행 시간 동안 지속되므로, 장기적인 데이터 보존이 가능함
• 단점
  • 메모리 누수: 메모리 해제를 제대로 처리하지 않으면 메모리 누수가 발생할 수 있음
  • 접근 시간: 힙은 트리 구조로 데이터에 접근하는 데에 추가적인 시간이 소요될 수 있음
  • 효율성: 동적할당 및 해제 작업은 상대적으로 오버헤드가 크므로, 스택에 비해 느릴 수 있음

# 파이썬 heapq 모듈을 이용한 힙 구현 예시
import heapq

heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 2)
heapq.heappush(heap, 7)

# 마지막 삽입한 값이 아닌 값 반환 가능 (heapq.heappop 함수는 힙에서 가장 작은 값을 삭제하고 반환)
print("힙에서 팝된 최소 항목:", heapq.heappop(heap))  # 출력: 2
print("힙에서 팝된 최소 항목:", heapq.heappop(heap))  # 출력: 5

FastAPI 개발자 tiangolo는 FastAPI를 어떻게 사용할까 궁금해서 찾아본 프로젝트

https://github.com/tiangolo/full-stack-fastapi-postgresql

이번 게시글에서는 DB 초기화를 어떻게 하는지 분석함

소스 개요

이 프로젝트에서는 prestart.sh을 이용하여 DB 초기화를 위한 스크립트 세 개를 수행함

#! /usr/bin/env bash

# Let the DB start
python /app/app/backend_pre_start.py # 데이터베이스 연결과 작동을 확인

# Run migrations
alembic upgrade head # 마이그레이션 수행

# Create initial data in DB
python /app/app/initial_data.py # DB 테이블에 초기 데이터 삽입

DB 초기화를 위한 스크립트 설명

backend_pre_start.py

이 스크립트는 데이터베이스가 정상적으로 작동하는지 확인하기 위해 사용될 수 있으며, 특히 Docker 컨테이너와 같은 환경에서 데이터베이스가 준비되었는지 확인할 때 유용함

init() 함수는 데이터베이스에 연결을 시도하고, 연결이 제대로 설정되었는지 확인

데이터베이스 연결이 실패할 경우 재시도할 수 있도록 @retry 데코레이터를 사용하여 init() 함수를 재시도 가능한 함수로 만듦

import logging

from sqlmodel import Session, select
from tenacity import after_log, before_log, retry, stop_after_attempt, wait_fixed

from app.core.db import engine

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

max_tries = 60 * 5  # 5 minutes
wait_seconds = 1


@retry(
    stop=stop_after_attempt(max_tries),
    wait=wait_fixed(wait_seconds),
    before=before_log(logger, logging.INFO),
    after=after_log(logger, logging.WARN),
)
def init() -> None:
    try:
        with Session(engine) as session:
            # Try to create session to check if DB is awake
            session.exec(select(1))
    except Exception as e:
        logger.error(e)
        raise e


def main() -> None:
    logger.info("Initializing service")
    init()
    logger.info("Service finished initializing")


if __name__ == "__main__":
    main()

alembic

데이터베이스 마이그레이션 도구로, DB에 테이블을 생성하거나 스키마 변경사항을 적용하는 데 사용

https://bigseok.tistory.com/entry/FastAPI-%EA%B0%9C%EB%B0%9C%EC%9E%90%EA%B0%80-%EC%A7%81%EC%A0%91-%EA%B0%9C%EB%B0%9C%ED%95%9C-FastAPI-backend-alembic

initial_data.py

이 스크립트는 데이터베이스 테이블에 초기 데이터를 삽입하는 모듈 init_db를 실행하여 데이터베이스를 초기화

init() 함수는 데이터베이스에 연결한 후, init_db() 함수를 사용하여 초기 데이터를 삽입

import logging

from sqlmodel import Session

from app.core.db import engine, init_db

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def init() -> None:
    with Session(engine) as session:
        init_db(session)


def main() -> None:
    logger.info("Creating initial data")
    init()
    logger.info("Initial data created")


if __name__ == "__main__":
    main()

app.core.db.py > init_db

# init_db
from sqlmodel import Session, create_engine, select

from app import crud
from app.core.config import settings
from app.models import User, UserCreate

engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))


# make sure all SQLModel models are imported (app.models) before initializing DB
# otherwise, SQLModel might fail to initialize relationships properly
# for more details: https://github.com/tiangolo/full-stack-fastapi-postgresql/issues/28


def init_db(session: Session) -> None:
    # Tables should be created with Alembic migrations
    # But if you don't want to use migrations, create
    # the tables un-commenting the next lines
    # from sqlmodel import SQLModel

    # from app.core.engine import engine
    # This works because the models are already imported and registered from app.models
    # SQLModel.metadata.create_all(engine)

    user = session.exec(
        select(User).where(User.email == settings.FIRST_SUPERUSER)
    ).first()
    if not user:
        user_in = UserCreate(
            email=settings.FIRST_SUPERUSER,
            password=settings.FIRST_SUPERUSER_PASSWORD,
            is_superuser=True,
        )
        user = crud.create_user(session=session, user_create=user_in)

참고 : FastAPI 개발자가 직접 개발한 FastAPI backend 시리즈

FastAPI 개발자가 직접 개발한 FastAPI backend 프로젝트 구조

FastAPI 개발자가 직접 개발한 FastAPI backend alembic

FastAPI 개발자가 직접 개발한 FastAPI backend DB 초기화 방법

FastAPI 개발자가 직접 개발한 FastAPI backend pyproject.toml

FastAPI 개발자가 직접 개발한 FastAPI backend 설정파일 config.py

FastAPI 개발자가 직접 개발한 FastAPI backend 데이터모델 models.py

파이썬에서 데이터를 다루는 데에 있어  Iterable, Iterator, Generator는 중요한 개념 중 하나임

이들을 적절히 활용하면 메모리를 효율적으로 활용하고, 코드를 간결하게 작성할 수 있음

1. Iterable(이터러블) 객체

이터러블(Iterable) 객체는 파이썬에서 하나씩 값을 반환할 수 있는 객체

리스트, 튜플, 문자열 등 파이썬에서 제공하는 대부분의 데이터 구조는 이터러블함

• 이터러블 객체의 특징
  • for 문과 같은 반복문에서 사용 가능
  • zip, map과 같은 함수에서도 사용 가능

2. Iterator(이터레이터) 객체

이터레이터(Iterator)는 이터러블 객체에서 값을 하나씩 가져오는 객체

이터레이터는 iter() 함수로 생성되며, __next__() 메서드 또는 내장 함수 next()를 사용하여 다음 값을 가져옴

• 이터레이터의 특징
  • 필요한 값만 생성하여 반환하므로 메모리 효율적
  • 모든 원소를 다 반환한 후에는 StopIteration 예외가 발생
• 이터레이터를 사용하면 효율적인 경우
  • 대용량 데이터 처리: 대용량의 데이터를 처리해야 할 때 이터레이터를 사용하면 필요한 값만 메모리에 로드하여 효율적으로 처리함
  • 순차적인 데이터 접근: 데이터를 순차적으로 접근해야 할 때 이터레이터를 사용하여 데이터를 하나씩 처리 가능
  • 데이터 순회: 데이터를 반복적으로 순회하면서 각각의 요소에 접근해야 할 때 유용

3. Generator(제너레이터)

제너레이터(Generator)는 값을 생성해 주는 함수

yield 키워드를 사용하여 값을 반환하면서 함수의 실행 상태를 유지

• 제너레이터의 특징
  • 코드가 간결해지며, 상태를 유지하면서 값을 생성 가능
  • 메모리 사용량이 적음
  • 무한한 시퀀스를 생성 가능
• 제너레이터를 사용하면 효율적인 경우
  • 대용량 데이터 생성: 대용량의 데이터를 생성해야 할 때 제너레이터를 사용하여 필요한 값만 생성할 수 있음
  • 지연 평가(Lazy Evaluation)가 필요한 경우: 값이 필요한 시점까지 연산을 미루고 싶을 때 제너레이터를 사용하여 지연 평가를 구현할 수 있음
  • 무한한 시퀀스 생성: 무한한 시퀀스를 생성해야 할 때 제너레이터를 사용하여 끝없는 데이터 스트림을 효과적으로 표현할 수 있음

4. 예제

# Iterable 객체 사용한 예제
for element in iterable_object:
    print(element)

# Iterable과 Iterator를 함께 사용한 예제
iterator_object = iter(iterable_object)
while True:
    try:
        element = next(iterator_object)
        print(element)
    except StopIteration:
        break

# Iterator를 사용한 예제
class MyIterator:
    def __init__(self, data):
        self.data = data
        self.index = 0

    def __iter__(self):
        return self

    def __next__(self):
        if self.index >= len(self.data):
            raise StopIteration
        value = self.data[self.index]
        self.index += 1
        return value

my_list = [1, 2, 3, 4, 5]
my_iterator = MyIterator(my_list)
for item in my_iterator:
    print(item)

# Generator를 사용한 예제
def generator_square(end):
    for number in range(0, end):
        yield number**2

square_results = generator_square(999999)
for result in square_results:
    print(result)

FastAPI 개발자 tiangolo는 FastAPI를 어떻게 사용할까 궁금해서 찾아본 프로젝트

https://github.com/tiangolo/full-stack-fastapi-postgresql

이번 게시글에서는 설정파일을 어떻게 구성하고 구조화하였는지 분석하려고 함

환경변수는 pydantic_settings를 기반으로 구성되어있으므로 pydantic_settings를 모른다면 pydantic_settings를 이용한 환경설정을 먼저 보고 오는 것을 권장함

소스 개요

config.py 파일 경로는 다음과 같음

```

app

└──  core # 데이터베이스 마이그레이션을 관리하는 Alembic 설정 파일 및 스크립트가 저장된 디렉토리

         └── config.py # 앱 전역 설정파일

```

import secrets
from typing import Any

from pydantic import (
    AnyHttpUrl,
    HttpUrl,
    PostgresDsn,
    ValidationInfo,
    field_validator,
)
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str = secrets.token_urlsafe(32)
    # 60 minutes * 24 hours * 8 days = 8 days
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
    SERVER_HOST: AnyHttpUrl
    # BACKEND_CORS_ORIGINS is a JSON-formatted list of origins
    # e.g: '["http://localhost", "http://localhost:4200", "http://localhost:3000", \
    # "http://localhost:8080", "http://local.dockertoolbox.tiangolo.com"]'
    BACKEND_CORS_ORIGINS: list[AnyHttpUrl] | str = []

    @field_validator("BACKEND_CORS_ORIGINS", mode="before")
    @classmethod
    def assemble_cors_origins(cls, v: str | list[str]) -> list[str] | str:
        if isinstance(v, str) and not v.startswith("["):
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, list | str):
            return v
        raise ValueError(v)

    PROJECT_NAME: str
    SENTRY_DSN: HttpUrl | None = None

    @field_validator("SENTRY_DSN", mode="before")
    @classmethod
    def sentry_dsn_can_be_blank(cls, v: str) -> str | None:
        if not v:
            return None
        return v

    POSTGRES_SERVER: str
    POSTGRES_USER: str
    POSTGRES_PASSWORD: str
    POSTGRES_DB: str
    SQLALCHEMY_DATABASE_URI: PostgresDsn | None = None

    @field_validator("SQLALCHEMY_DATABASE_URI", mode="before")
    def assemble_db_connection(cls, v: str | None, info: ValidationInfo) -> Any:
        if isinstance(v, str):
            return v
        return PostgresDsn.build(
            scheme="postgresql+psycopg",
            username=info.data.get("POSTGRES_USER"),
            password=info.data.get("POSTGRES_PASSWORD"),
            host=info.data.get("POSTGRES_SERVER"),
            path=f"{info.data.get('POSTGRES_DB') or ''}",
        )

    SMTP_TLS: bool = True
    SMTP_PORT: int | None = None
    SMTP_HOST: str | None = None
    SMTP_USER: str | None = None
    SMTP_PASSWORD: str | None = None
    # TODO: update type to EmailStr when sqlmodel supports it
    EMAILS_FROM_EMAIL: str | None = None
    EMAILS_FROM_NAME: str | None = None

    @field_validator("EMAILS_FROM_NAME")
    def get_project_name(cls, v: str | None, info: ValidationInfo) -> str:
        if not v:
            return info.data["PROJECT_NAME"]
        return v

    EMAIL_RESET_TOKEN_EXPIRE_HOURS: int = 48
    EMAIL_TEMPLATES_DIR: str = "/app/app/email-templates/build"
    EMAILS_ENABLED: bool = False

    @field_validator("EMAILS_ENABLED", mode="before")
    def get_emails_enabled(cls, v: bool, info: ValidationInfo) -> bool:
        return bool(
            info.data.get("SMTP_HOST")
            and info.data.get("SMTP_PORT")
            and info.data.get("EMAILS_FROM_EMAIL")
        )

    # TODO: update type to EmailStr when sqlmodel supports it
    EMAIL_TEST_USER: str = "test@example.com"
    # TODO: update type to EmailStr when sqlmodel supports it
    FIRST_SUPERUSER: str
    FIRST_SUPERUSER_PASSWORD: str
    USERS_OPEN_REGISTRATION: bool = False
    model_config = SettingsConfigDict(case_sensitive=True)


settings = Settings()

주요 설정

• 일반 설정
  • API_V1_STR: API 버전을 지정하는 문자열
  • SECRET_KEY: 인증 토큰 및 세션 등을 위한 시크릿 키
  • ACCESS_TOKEN_EXPIRE_MINUTES: 액세스 토큰 만료 시간(분 단위)
  • PROJECT_NAME: 프로젝트의 이름
  • EMAIL_RESET_TOKEN_EXPIRE_HOURS: 이메일 리셋 토큰의 만료 시간(시간 단위)
  • EMAIL_TEMPLATES_DIR: 이메일 템플릿 디렉토리 경로
  • EMAILS_ENABLED: 이메일 전송 기능 활성화 여부
  • EMAIL_TEST_USER: 이메일 테스트 사용자
  • FIRST_SUPERUSER: 최초 슈퍼 사용자 이름
  • FIRST_SUPERUSER_PASSWORD: 최초 슈퍼 사용자 비밀번호
  • USERS_OPEN_REGISTRATION: 사용자의 개방된 가입 여부
• CORS 관련 설정
  • BACKEND_CORS_ORIGINS: 허용된 CORS 원본 목록.
• 데이터베이스 관련 설정
  • POSTGRES_SERVER: PostgreSQL 서버 호스트
  • POSTGRES_USER: PostgreSQL 사용자 이름
  • POSTGRES_PASSWORD: PostgreSQL 비밀번호
  • POSTGRES_DB: PostgreSQL 데이터베이스 이름
  • SQLALCHEMY_DATABASE_URI: SQLAlchemy를 위한 PostgreSQL 데이터베이스 URI
• 이메일 관련 설정
  • SMTP_TLS: SMTP TLS 활성화 여부
  • SMTP_PORT: SMTP 포트
  • SMTP_HOST: SMTP 호스트
  • SMTP_USER: SMTP 사용자
  • SMTP_PASSWORD: SMTP 비밀번호
  • EMAILS_FROM_EMAIL: 이메일 발신자 이메일 주소
  • EMAILS_FROM_NAME: 이메일 발신자 이름
• 서버 관련 설정
  • SERVER_HOST: 서버 호스트 URL
• 로그 및 모니터링 관련 설정
  • SENTRY_DSN: Sentry 프로젝트 DSN

소스 분석

Python에서 환경 설정을 다루는 데 사용되는 Pydantic 라이브러리를 사용하여 구조화함

Pydantic 및 pydantic_settings 라이브러리를 통해 설정 값을 관리하고 유효성을 검사하는 데 사용

Pydantic을 사용하면 설정을 타입 안정성이 있는 Python 객체로 정의할 수 있음

설정 파일에 있는 일부 필드들은 다른 필드의 값에 의존하는데, 이를 위하여 @field_validator 데코레이터와 ValidationInfo 객체를 이용

위의 코드를 보면 EMAILS_FROM_NAME 필드가 PROJECT_NAME 필드의 값을 기본값으로 사용하도록 설정되어 있음

이를 위해 ValidationInfo 객체를 사용하여 필요한 정보를 얻고, 필드 값을 조작하여 반환하고 있음

타입 힌트(from typing import Any) 및 Pydantic의 타입들(AnyHttpUrl, PostgresDsn, ValidationInfo)을 사용하여 데이터의 형식을 지정하고 유효성을 검사하는 데 사용

참고

Fastapi 설정 docs : https://fastapi.tiangolo.com/ru/advanced/settings/

FastAPI 개발자가 직접 개발한 FastAPI backend 프로젝트 구조

FastAPI 개발자가 직접 개발한 FastAPI backend alembic

FastAPI 개발자가 직접 개발한 FastAPI backend DB 초기화 방법

FastAPI 개발자가 직접 개발한 FastAPI backend pyproject.toml

FastAPI 개발자가 직접 개발한 FastAPI backend 설정파일 config.py

FastAPI 개발자가 직접 개발한 FastAPI backend 데이터모델 models.py

BaseSettings

BaseSettings는 Pydantic에서 설정을 정의하기 위한 기본 클래스

기본적으로 이름이 지정된 환경 변수를 읽어와서 설정값을 자동으로 설정

이 클래스는 Pydantic 라이브러리의 일부로서 설정값의 유효성을 검사하고 기본값을 지정할 수 있는 기능을 제공

from pydantic import BaseSettings

class Settings(BaseSettings):
    DATABASE_URL: str
    API_KEY: str = "default_key"

settings = Settings()

BaseSettings는 설정 초기화 시 여러가지 기본 설정들이 있음

 예를 들어 기본적으로 대소문자 구분을 하지 않아 설정 클래스의 필드명이 data_base_url 이라면, DATABASE_URL이라는 이름의 환경 변수를 자동으로 찾아 해당 값을 읽어와 설정

또는 환경변수를 파일로 관리하는 경우 환경변수 파일 경로를 설정할 필요가 있음

이런 기본 설정은 SettingsConfigDict를 이용하여 설정 가능함 (하단 SettingConfigDict 참고)

SettingsConfigDict

SettingsConfigDict는 설정을 딕셔너리 형태로 관리하기 위한 클래스로, 설정 초기화 옵션을 재정의 하는 데 사용

일반적으로 BaseSettings와 함께 사용하며,  BaseSettings를 상속받은 설정은 자동으로 SettingsConfigDict가 인식

from pydantic_settings import BaseSettings, SettingsConfigDict

import os
os.environ["DATABASE_URL"] = "db_url"
os.environ["TEST_DATABASE_URL"] = "test_db_url"
os.environ["API_KEY"] = "default"
os.environ["TEST_API_KEY"] = "test"

class Settings(BaseSettings):
    DATABASE_URL: str
    API_KEY: str = "lower_case"
    model_config = SettingsConfigDict(
        case_sensitive=True, 
        env_prefix='TEST_', 
        env_file='.env', 
        env_file_encoding='utf-8')

settings = Settings()

print(settings.DATABASE_URL) # result : test_db_url
print(settings.API_KEY) # result : test

• case_sensitive : 설정을 딕셔너리로 관리할 때 대소문자를 구분하도록 설정하는 옵션 (기본값 : False)
• validate_default : 필드 기본값에 대한 밸리데이션 여부를 설정하는 옵션 (기본값 : False)
• env_prefix : 환경변수의 접두사를 설정 가능  (기본값 : null)
• env_file : 환경변수 설정파일 지정하는 옵션
• env_file_encoding : 환경변수 설정파일 인코딩 지정하는 옵션

참고

Pydantic Settings 설정 방법 docs : https://docs.pydantic.dev/latest/concepts/pydantic_settings/

다양한 프레임워크와 ORM 도구에서 데이터 모델을 정의하는데 사용되는 파일로, 데이터베이스 테이블 구조를 코드로 표현

주 구성요소

• 테이블을 나타내는 클래스 정의
• 각 클래스 내에 속성 정의 (테이블 컬럼 정의)
• 테이블 간의 관계 정의 (Foreign key, one to one, many to many 등)
• 모델의 메타데이터 정의 
• 필요한 인덱스, 제약조건 등 정의
• 모델 검증을 위한 로직 (잘못된 입력 방지 및 특정 규칙 적용)
• 데이터베이스 마이그레이션을 위한 설정

from sqlmodel import Field, Relationship, SQLModel


# Shared properties
# TODO replace email str with EmailStr when sqlmodel supports it
class UserBase(SQLModel):
    email: str = Field(unique=True, index=True)
    is_active: bool = True
    is_superuser: bool = False
    full_name: str | None = None


# Properties to receive via API on creation
class UserCreate(UserBase):
    password: str


# TODO replace email str with EmailStr when sqlmodel supports it
class UserCreateOpen(SQLModel):
    email: str
    password: str
    full_name: str | None = None


# Properties to receive via API on update, all are optional
# TODO replace email str with EmailStr when sqlmodel supports it
class UserUpdate(UserBase):
    email: str | None = None
    password: str | None = None


# TODO replace email str with EmailStr when sqlmodel supports it
class UserUpdateMe(SQLModel):
    full_name: str | None = None
    email: str | None = None


class UpdatePassword(SQLModel):
    current_password: str
    new_password: str


# Database model, database table inferred from class name
class User(UserBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    hashed_password: str
    items: list["Item"] = Relationship(back_populates="owner")


# Properties to return via API, id is always required
class UserOut(UserBase):
    id: int


class UsersOut(SQLModel):
    data: list[UserOut]
    count: int


# Shared properties
class ItemBase(SQLModel):
    title: str
    description: str | None = None


# Properties to receive on item creation
class ItemCreate(ItemBase):
    title: str


# Properties to receive on item update
class ItemUpdate(ItemBase):
    title: str | None = None


# Database model, database table inferred from class name
class Item(ItemBase, table=True):
    id: int | None = Field(default=None, primary_key=True)
    title: str
    owner_id: int | None = Field(default=None, foreign_key="user.id", nullable=False)
    owner: User | None = Relationship(back_populates="items")


# Properties to return via API, id is always required
class ItemOut(ItemBase):
    id: int
    owner_id: int


class ItemsOut(SQLModel):
    data: list[ItemOut]
    count: int


# Generic message
class Message(SQLModel):
    message: str


# JSON payload containing access token
class Token(SQLModel):
    access_token: str
    token_type: str = "bearer"


# Contents of JWT token
class TokenPayload(SQLModel):
    sub: int | None = None


class NewPassword(SQLModel):
    token: str
    new_password: str

특징 및 장점

• 코드가 간결하며, Pydantic 및 SQLModel을 사용하여 모델을 정의하고 데이터베이스 테이블을 생성
• UserBase를 기반으로 하는 다양한 파생 클래스들이 있는데 공통된 속성을 하위 클래스에서 확장하여 사용하는 방식이 효과적이며, 코드 재사용성을 높임
• Pydantic 모델은 FastAPI에서 사용하기에 이상적이며, API의 요청 및 응답에 대한 유효성 검사 및 자동 문서화를 제공
• 모델 정의에서 데이터베이스 테이블 정의까지 일관된 패턴을 사용하고 있어 코드의 이해와 유지보수에 유리
• UsersOut 및 ItemsOut과 같은 응답 모델을 정의하여 API 응답에 대한 구조를 명확하게 나타냄

주의할 점

• 클래스 이름을 기반으로 한 암시적인 테이블 이름이 사용되고 있음 (명시적인 테이블 이름 정의가 필요할 수 있음)
• 모델 정의에 클래스 계층 구조가 있어 데이터베이스와 API 간의 복잡한 관계에서는 코드의 이해가 어려워질 수 있음
• SQLModel이 EmailStr를 지원하지 않을 때까지 email 속성은 일반 문자열로 남아있음 (주석에도 명시됨)
• owner_id와 user.id 간의 Foreign Key 관계가 암시적으로 정의되어 있음 (명시적인 Foreign Key 제약이 필요할 수 있음)

참고 : FastAPI 개발자가 직접 개발한 FastAPI backend 시리즈

FastAPI 개발자가 직접 개발한 FastAPI backend 프로젝트 구조

FastAPI 개발자가 직접 개발한 FastAPI backend alembic

FastAPI 개발자가 직접 개발한 FastAPI backend DB 초기화 방법

FastAPI 개발자가 직접 개발한 FastAPI backend pyproject.toml

FastAPI 개발자가 직접 개발한 FastAPI backend 설정파일 config.py

FastAPI 개발자가 직접 개발한 FastAPI backend 데이터모델 models.py

FastAPI 개발자 tiangolo는 FastAPI를 어떻게 사용할까 궁금해서 찾아본 프로젝트

https://github.com/tiangolo/full-stack-fastapi-postgresql

이번 게시글에서는 프로젝트 데이터베이스 마이그레이션 도구인 alembic을 어떻게 적용하였는지 분석하려고 함

alembic 관련 파일은 아래와 같음

```

.env # 환경 변수 설정이 담긴 파일 → DB 설정을 이 파일에서 확인 가능

alembic.ini # Alembic의 설정 파일로, 데이터베이스 마이그레이션 설정을 정의 

app

└──  alembic # 데이터베이스 마이그레이션을 관리하는 Alembic 설정 파일 및 스크립트가 저장된 디렉토리

           ├──  env.py # alembic 실행 시 실행되는 파일로, SQLAIchemy의 Engine을 설정하고 마이그레이션 스크립트 실행
           └──  versions # 버전 별 데이터베이스 마이그레이션 스크립트가 저장된 디렉토리

                   └── e2412789c190_initialize_models.py # 데이터 베이스 테이블 및 인덱스 생성 스크립트

└──  models.py # ORM 도구에서 데이터 모델을 정의하는데 사용되는 파일로, 이 파일의 모델과 DB 차이를 바탕으로 마이그레이션 스크립트가 생성됨

```

.env

파이썬에서는 프로젝트 별로 환경변수를 관리하기 위하여 .env파일을 널리 사용함

.env파일에 설정된 alembic에 사용되는 환경변수는 아래와 같음

# Postgres
POSTGRES_SERVER=db
POSTGRES_USER=postgres
POSTGRES_DB=app
POSTGRES_PASSWORD=changethis

alembic.ini

alembic.ini 파일은 Alembic 데이터베이스 마이그레이션 도구의 설정 파일로, 주로 데이터베이스 연결 및 마이그레이션 관련 설정을 포함

• 일반적으로 사용되는 설정
  • 데이터베이스 연결 정보 설정 (sqlalchemy.url)
  • 마이그레이션 스크립트 경로 설정 (script_location)
  • 로깅 설정 (loggers, handlers, formatters)
  • 버전 관리 설정 (version_table)
  • 그 외 커스텀 설정

tiangolo 백엔드 프로젝트의 alembic.ini는 아래와 같은 설정으로 구성되어 있음

# A generic, single database configuration.

[alembic]
script_location = app/alembic # 마이그레이션 스크립트가 있는 디렉토리의 경로를 지정

[loggers]
keys = root,sqlalchemy,alembic # 설정될 로거의 이름을 지정

[handlers]
keys = console # "console"이라는 핸들러가 설정되어 있다는 것을 지정

[formatters]
keys = generic # "generic"이라는 포매터가 설정되어 있다는 것을 지정

[logger_root] # "root"에 대한 로거를 구성
level = WARN
handlers = console
qualname = 

[logger_sqlalchemy] # "sqlalchemy"에 대한 로거를 구성
level = WARN 
handlers =
qualname = sqlalchemy.engine

[logger_alembic] # "alembic"에 대한 로거를 구성
level = INFO
handlers =
qualname = alembic

[handler_console] # 콘솔 핸들러를 구성 (콘솔 핸들러는 로그 메시지를 콘솔에 출력)
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic] #  "generic" 포매터에 대한 설정을 정의
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S

env.py

 alembic 실행 시 실행되는 파일로, SQLAIchemy의 Engine을 설정하고 마이그레이션 스크립트 실행

• 주요 작업
  • 모듈 및 패키지 가져오기
  • context.config를 사용하여 Alembic Config 객체를 설정
    이 객체는 사용 중인 .ini 파일 내의 값을 제공
  • fileConfig(config.config_file_name)을 호출하여 Python 로깅을 설정
    .ini 파일의 로깅 구성을 기반으로 로그 기능을 활성화
  • app.models에서 SQLModel을 가져와 target_metadata로 설정
    마이그레이션 스크립트에서 사용할 데이터베이스 모델의 메타데이터를 지정
    
  • get_url함수는 .env(환경 변수)를 통해 PostgreSQL 데이터베이스 연결 URL을 생성
    
  • run_migrations_offline 함수는 오프라인 모드에서 마이그레이션을 실행
  • run_migrations_online 함수는 온라인 모드에서 마이그레이션 실행
  • context.is_offline_mode()를 사용하여 현재 마이그레이션 모드가 오프라인인지 확인하고, 그에 따라 run_migrations_offline 또는 run_migrations_online 함수를 호출하여 마이그레이션을 실행

versions / e2412789c190_initialize_models.py

데이터 베이스 테이블 및 인덱스 생성 스크립트

일반적으로 alembic을 이용하여 마이그레이션 스크립트를 자동적으로 생성하면 models.py파일을 기반으로 테이블 생성/삭제 스크립트만 생성됨

index 생성 스크립트는 수동으로 추가했을 것으로 예상됨

alembic이 뭔지 모르는 경우 alembic 소개 및 기초 사용법 포스팅 참고

"""Initialize models

Revision ID: e2412789c190
Revises:
Create Date: 2023-11-24 22:55:43.195942

"""
import sqlalchemy as sa
import sqlmodel.sql.sqltypes
from alembic import op

# revision identifiers, used by Alembic.
revision = "e2412789c190"
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "user",
        sa.Column("email", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
        sa.Column("is_active", sa.Boolean(), nullable=False),
        sa.Column("is_superuser", sa.Boolean(), nullable=False),
        sa.Column("full_name", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column(
            "hashed_password", sqlmodel.sql.sqltypes.AutoString(), nullable=False
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
    op.create_table(
        "item",
        sa.Column("description", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("title", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
        sa.Column("owner_id", sa.Integer(), nullable=False),
        sa.ForeignKeyConstraint(
            ["owner_id"],
            ["user.id"],
        ),
        sa.PrimaryKeyConstraint("id"),
    )
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table("item")
    op.drop_index(op.f("ix_user_email"), table_name="user")
    op.drop_table("user")
    # ### end Alembic commands ###

참고 : FastAPI 개발자가 직접 개발한 FastAPI backend 시리즈

FastAPI 개발자가 직접 개발한 FastAPI backend 프로젝트 구조

FastAPI 개발자가 직접 개발한 FastAPI backend alembic

FastAPI 개발자가 직접 개발한 FastAPI backend DB 초기화 방법

FastAPI 개발자가 직접 개발한 FastAPI backend pyproject.toml

FastAPI 개발자가 직접 개발한 FastAPI backend 설정파일 config.py

FastAPI 개발자가 직접 개발한 FastAPI backend 데이터모델 models.py

Flask-Migrate는 db와 sqlalchemy models.py의 차이를 추적하여 업그레이드/다운그레이드 스크립트 자동으로 생성해줌

만약에 서비스에 필요한 DB table 초기 데이터도 함께 관리하고 싶다면 마이그레이션 스크립트에 수동으로  추가해주어야 함

1. DB 초기화

flask db init

위의 명령어를 수행한 위치에 migrations 폴더 자동으로 생성됨

2. 마이그레이션 스크립트 생성

flask db migrate -m "Initial migration."

migrations/versions 폴더 하위에 {revision}_{message}.py 스크립트가 자동으로 생성됨

3. 마이그레이션 스크립트에 수동으로 초기데이터 변경내용 추가

마이그레이션 스크립트 업그레이드 함수에 초기데이터삽입 코드 추가예시

# /migrations/versions/{revision}_{message}.py
def upgrade():
    ...생략...
    # ### end Alembic commands ###
    add_initial_data()

def add_initial_data():
    op.bulk_insert(
        sa.table('tb_users',
            sa.column('name'),
            sa.column('user_id'),
            sa.column('password'),
            sa.column('admin_yn'),
            sa.column('email')
        ),
        [
            {
                'name': '관리자',
                'user_id': 'admin',
                'password': 'admin123!@#',
                'admin_yn': 'Y',
                'email': 'admin@test.com'
            }
        ]
    )

• 💡 Alembic에서 op 객체(operations의 약자)는 다양한 데이터베이스 스키마 작업을 수행할 수 있는 함수 집합을 제공 https://alembic.sqlalchemy.org/en/latest/ops.html

• alembic op모듈 주요 기능
  1. 테이블 작업
     • op.create_table(): 새로운 테이블을 생성
     • op.drop_table(): 기존 테이블을 삭제
     • op.add_column(): 기존 테이블에 새로운 열을 추가
     • op.drop_column(): 기존 테이블에서 열을 삭제
     • op.alter_column(): 열의 속성을 변경
  2. 인덱스 작업
     • op.create_index(): 새로운 인덱스를 생성
     • op.drop_index(): 기존 인덱스를 삭제
  3. 데이터 작업
     • op.execute(): SQL 문을 직접 실행
  4. 제약 조건 작업
     • op.create_foreign_key(): 새로운 외래 키 제약 조건을 생성
     • op.create_primary_key(): 새로운 기본 키 제약 조건을 생성
     • op.create_check_constraint(): 새로운 체크 제약 조건을 생성
  5. 기타 작업
     • op.rename_table(): 기존 테이블의 이름을 변경
     • op.rename_column(): 기존 열의 이름을 변경
     • op.create_sequence(): 새로운 데이터베이스 시퀀스를 생성
  6. 대량 데이터 작업
     • op.bulk_insert(): 테이블에 대량 삽입을 수행
     • op.bulk_update(): 테이블에서 대량 업데이트를 수행

Flask-Migrate란?

Flask-Migrate는 SQLAlchemy를 사용하는 Flask 애플리케이션에서 데이터베이스 스키마를 효과적으로 관리하기 위한 extension

Flask-Migrate는 Alembic이라는 데이터베이스 마이그레이션 툴을 내부적으로 사용

Flask 애플리케이션과의 원활한 통합을 제공하여 데이터베이스 마이그레이션을 더욱 편리하게 관리 가능

https://flask-migrate.readthedocs.io/en/latest/#why-use-flask-migrate-vs-alembic-directly

Flask-Migrate 사용법

설치

pip install Flask-Migrate

환경변수 설정

flask app 실행파일명이 app.py나 wsgi.py가 아닌 이상 환경변수(.env파일)에 flask app 실행파일명 설정 필수

FLASK_APP= "__run__.py"

마이그레이션 (alembic) 초기화

flask db init

마이그레이션 스크립트 생성

flask db migrate -m "Initial migration."

현재의 SQLAlchemy 모델과 데이터베이스 상태를 비교하여 마이그레이션 스크립트를 자동으로 생성

버전업 (DB에 스키마 변경사항 반영)

flask db upgrade [--sql] [--tag TAG] <revision>
# 데이터베이스를 업그레이드 revision주어지지 않으면 최신버전

버전 다운그레이드 (최초 버전으로)

flask db downgrade [--sql] [--tag TAG] <revision>
# 데이터베이스를 다운그레이드, revision이 주어지지 않으면 -1