app/main.py

from fastapi import FastAPI

app = FastAPI(
    title="Holiday Keeper API",
    description="공휴일 정보 관리 API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

@app.get("/")
async def root():
    return {"message": "Holiday Keeper API is running!"}

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "service": "holiday-keeper-api",
        "version": "1.0.0"
    }

@app.get("/test")
async def test():
    return {
        "message": "테스트 성공",
        "data": ["공휴일1", "공휴일2", "공휴일3"]
    }

• 서버 실행 테스트

  # 스크립트 사용
  ./scripts/dev.sh
  
  # 직접 서버 실행
  poetry run uvicorn app.main:app --reload --host 127.0.0.1 --port 8090
  

  • 확인
    • http://localhost:8090 접속
    • http://localhost:8090/docs 접속 (Swagger)

Spring Boot (자동 구성)

• application.yml

  server:
  port: 8090
  spring:
  datasource:
      url: jdbc:postgresql://localhost:5432/holiday_db
  

  • 설정 값 저장 (DB 연결 정보, 포트 등)
  • Spring Boot의 핵심 중 하나인 Auto Configuration으로 자동 인식
  
  
  

• WebConfig.java

  @Configuration
  public class WebConfig implements WebMvcConfigurer {
      
      @Override
      public void addCorsMappings(CorsRegistry registry) {
          registry.addMapping("/**")
              .allowedOrigins("http://localhost:3000")
              .allowedMethods("*");
      }
      
      @Override
      public void addInterceptors(InterceptorRegistry registry) {
          registry.addInterceptor(new LoggingInterceptor());
      }
  }
  

  • 설정 값 적용, CORS 직접 설정, 인터셉터 등록
  • @Configuration으로 자동 스캔
  
  
  

• LoggingInterceptor.java

  @Component
  public class LoggingInterceptor implements HandlerInterceptor {
      @Override
      public boolean preHandle(HttpServletRequest request, 
                          HttpServletResponse response, Object handler) {
          log.info("Request: {} {}", request.getMethod(), request.getRequestURI());
          return true;
      }
  }
  

  • 실제 로직 구현

FastAPI (수동 제어)

• .env

  SERVER_HOST="0.0.0.0"
  SERVER_PORT=8090
  DATABASE_URL="postgresql://localhost:5432/holiday_db"
  CORS_ORIGINS="http://localhost:3000,http://localhost:5173"
  

  • 설정 값 저장 (DB 연결 정보, 포트 등)
  
  
  

• config.py

  from pydantic_settings import BaseSettings
  
  class Settings(BaseSettings):
      server_host: str = "0.0.0.0"
      server_port: int = 8090
      database_url: str = "postgresql://..."
      cors_origins: str = "http://localhost:3000"
      
      model_config = {"env_file": ".env"}
      
      @property
      def cors_origins_list(self) -> List[str]:
          return self.cors_origins.split(",")
  
  settings = Settings()
  

  • .env 파일 읽기
  • 설정 객체 생성
  
  
  

• main.py

  from fastapi import FastAPI
  from app.core.config import settings
  from app.core.middleware import add_middleware
  
  app = FastAPI(title="Holiday Keeper API")
  
  add_middleware(app)
  
  @app.get("/")
  async def root():
      return {"message": "Hello"}
  

  • 수동으로 모든 설정 연결
  
  
  

• middleware.py

  from fastapi.middleware.cors import CORSMiddleware
  from app.core.config import settings
  
  def add_middleware(app: FastAPI):
      # CORS 설정 (WebConfig 역할)
      app.add_middleware(
          CORSMiddleware,
          allow_origins=settings.cors_origins_list,
          allow_credentials=True,
          allow_methods=["*"]
      )
      
      # 로깅 미들웨어 (Interceptor 역할)
      @app.middleware("http")
      async def log_requests(request, call_next):
          start_time = time.time()
          response = await call_next(request)
          process_time = time.time() - start_time
          logger.info(f"{request.method} {request.url} - {process_time:.3f}s")
          return response
  

  • Spring Boot의 WebConfig + Interceptor 구현을 한 파일에 합친 형태
  
  
  

• deps.py

  from fastapi import Depends
  from app.core.config import settings
  
  async def get_db():
      # DB 세션 의존성 주입용
      pass
  
  def get_current_user():
      # 인증 의존성 주입용  
      pass
  

  • Spring Boot의 @Autowired 의존성 주입을 수동으로 구현하는 곳

app/core/config.py

from functools import lru_cache
from typing import List
import os
from pathlib import Path

from pydantic import Field
from pydantic_settings import BaseSettings

environment = os.getenv("ENVIRONMENT", "dev")
env_file = f".env.{environment}"

_secrets_dir = Path("/run/secrets") if Path("/run/secrets").exists() else Path(os.getenv("SECRETS_PATH", "./secrets"))

class Settings(BaseSettings):
    project_name: str = Field(default="Holiday-keeper-fastapi")
    version: str = Field(default="1.0.0")
    environment: str = environment

    server_host: str = Field(default="0.0.0.0")
    server_port: int = Field(default=8090)
    cors_origins: str = Field(default="http://localhost:3000,http://localhost:5173")
    log_level: str = Field(default="INFO")

    redis_url: str = Field(default="redis://localhost:6379")
    nager_api_base_url: str = Field(default="https://date.nager.at/api/v3")

    database_url: str
    sync_database_url: str

    model_config = {
        "env_file": env_file,
        "env_file_encoding": "utf-8",
        "case_sensitive": False,
        "extra": "ignore",
        "secrets_dir": _secrets_dir,
        # "validate_default": True,
        # "use_enum_values": True
    }

    @property
    def cors_origins_list(self) -> List[str]:
        return [origin.strip() for origin in self.cors_origins.split(",")]

    @property
    def is_dev(self) -> bool:
        return self.environment.lower() == "dev"

@lru_cache()
def get_settings() -> Settings:
    return Settings()

settings = get_settings()

@lru_cache와 싱글톤 패턴

@lru_cache()
def get_settings():
    return Settings()

settings = get_settings()

• Spring의 경우 기본적으로 싱글톤이지만 FastAPI는 그렇지 않다. 객체를 Spring의 Bean 처럼 활용하기 위함이다

  
  

• get_settings( )

  • 단순한 팩토리 함수 (객체 생성 반환)
  
  
  

• @lru_cache

  • 함수 결과 캐싱 ⇒ 싱글톤 구현
  • 첫번째 호출 시에만 Settings() 객체 생성
  • 함수를 여러 번 호출해도 캐싱한 결과 반환
  • 여기까지가 Spring Boot의 싱글톤 빈을 수동으로 구현한 것이다.
  
  
  

• settings = get_settings( )

  • 전역 객체 생성
    • 모듈 로드 시 한 번만 실행
  • 다른 파일에서 from app.core.config import settings 시 같은 객체를 사용한다.
  • 이렇게 편의 객체를 만들지 않으면 매번 다른 파일에서 함수를 호출해야 한다

Pydantic의 Field

from pydantic import Field

class Settings(BaseSettings):

database_url: str

    server_port: int = Field(default=8090, ge=1024, le=65535, description="서버 포트")

• Field 함수를 사용하지 않는 경우

  • 예시 : database_url: str
  • 단순히 타입 힌트를 제공하며, 필드가 필수값임을 선언하는 것
  • BaseSettings는 이 필드 이름과 일치하는 것을 찾아 로딩한다.
    1. 환경 변수 (OS 환경변수)
    2. .env 파일 값
    3. model_config에 설정된 secrets_dir 내의 동일한 파일 이름
       • 예시 : database_url.txt
  
  

• Field 함수를 사용하는 경우

  • 단순히 타입을 선언하는 것이 아닌 기본값을 지정하거나 유효성 검사 등 설정 가능

  • Spring Boot의 @Value + Bean Validation을 합친 기능

  • 필드의 이름과 환경 변수의 이름이 같아야 동작한다
    

  • default

    • 기본값 설정
    • default 우선 순위
      1. 환경변수 (OS 환경변수)
      2. .env 파일 값
      3. Field(default=값)

  • ge/le

    • 범위 제한 (greater equal, less equal)
    
    

  • description

    • API 문서에 표시될 설명
    
    

  • alias

    # .env  
    MY_CUSTOM_PORT=8080
    
    # Python
    server_port: int = Field(default=8090, alias="MY_CUSTOM_PORT")
    

    • 다른 이름으로 매핑해야 하는 경우 사용

BaseSettings

from pydantic_settings import BaseSettings

environment = os.getenv("ENVIRONMENT", "dev")
env_file = f".env.{environment}"

_secrets_dir = Path("/run/secrets") if Path("/run/secrets").exists() else Path(os.getenv("SECRETS_PATH", "./secrets"))

class Settings(BaseSettings):
    name: str
    version: str = Field(default="1.0.0")
    
        model_config = {
            "env_file": env_file,            # 읽을 파일 경로
            "env_file_encoding": "utf-8",    # 파일 인코딩
            "case_sensitive": False,         # 대소문자 구분 안 함
            "extra": "ignore",               # 추가 필드 처리 방식
            "validate_default": True,        # 기본값도 검증할지
            "use_enum_values": True,         # Enum 값 사용
            "secrets_dir": _secrets_dir,     # Secret 폴더 경로 지정
        }

• BaseSettings

  • Spring Boot의 @ConfigurationProperties
  • .env 파일 자동 읽기
  • 환경변수 자동 매핑
  • 타입 변환 (문자열 → 정수, 불린 등)
  • 유효성 검증
  
  
  

• 프로파일 동적 설정

  environment = os.getenv("ENVIRONMENT", "dev")
  env_file = f".env.{environment}"
  
  class Settings(BaseSettings):
      model_config = {
                  "env_file": env_file,
      }
  

  • 환경 설정 파일
    • .env.dev
    • .env.prod
    • .env.test
  • 환경 변수 설정에 따라 동적으로 사용하는 설정이다.
  • 시크릿 설정은 시크릿 파일로 관리한다.
  
  
  

• _secrets_dir 전역 변수

  • Docker Secrets 환경의 경로인 /run/secrets를 먼저 확인하고 없다면, 로컬 환경을 확인한다.
  • 전역 변수로 설정하는 이유는 Settings 클래스 정의 시 그 값을 활용하기 위함이다
  
  
  

• model_config = { }

  • BaseSettings가 어떻게 환경변수를 읽을지 설정하는 메타데이터 설정

  • Settings 클래스가 정의될 때, model_config 내부의 값들도 함께 정해진다

  • Settings 클래스의 인스턴스가 생기기 전에는 클래스 내부 필드의 값을 참조할 수 없다
    

    • case_sensitive

      • Spring Boot는 kebab-case를 camelCase로 자동 변환
      • 환경변수 이름의 대소문자를 구분하는지 여부
      
      

    • extra: "ignore"

      • Spring Boot의 ignoreUnknownFields = true
      • Settings 클래스에 없는 필드에 대한 처리 여부
      • extra: "ignore"
        • Settings 클래스에 없는 환경변수 무시
        • extra: "allow"일 때만 settings.UNKNOWN_FIELD로 접근 가능
      
      

    • validate_default

      • 기본값이 잘못된 경우를 검증하는지 여부
      
      

    • use_enum_values

      • Settings 클래스에서 다른 enum 클래스를 필드 타입으로 사용할지 여부

        class Settings(BaseSettings):
                environment: Environment = Environment.DEV
        
        class Environment(Enum):
            DEV = "development"
            PROD = "production"
        

      • use_enum_values=True ⇒ "development" 값 사용

      • use_enum_values=False ⇒ Environment.DEV 객체 사용

      
      

    • secrets_dir

      • Secret 폴더의 경로를 지정하여 해당 경로의 값들도 읽을 수 있게 설정한다
  
  
  

• Pydantic v1 (구버전)

  class Settings(BaseSettings):
      name: str
      
      class Config:               # 내부 클래스
          env_file = ".env"
          case_sensitive = True
  

@property

class Settings:
    cors_origins: str = "http://localhost:3000,http://localhost:5173"
    
    @property
    def cors_origins_list(self) -> List[str]:
        return self.cors_origins.split(",")
        

# 사용법
settings.cors_origins_list

• 함수를 속성처럼 접근 가능하게 한다
• Java의 Getter와 같지만 호출 시 괄호가 필요 없다
• @{property이름}.setter

  class Settings:
      def __init__(self):
          self._cors_data = "http://localhost:3000"
      
      @property
      def cors_origins(self):  # property 이름
          return self._cors_data
      
      @cors_origins.setter
      def cors_origins(self, value):
          self._cors_data = value
  

  • 이렇게 setter를 지정할 수 있다.
  • 단, 메서드 이름도 property와 동일해야 한다.
  • 환경변수 설정이므로 Settings 클래스에서는 쓸 일이 없다.

MySQL 연결 설정

# secrets/database_url
mysql+aiomysql://root:1234@127.0.0.1:3306/hk_db?charset=utf8mb4

# secrets/sync_database_url
mysql+pymysql://root:1234@127.0.0.1:3306/hk_db?charset=utf8mb4

# .env.dev
DATABASE_URL="mysql+aiomysql://root:1234@127.0.0.1:3306/hk_db?charset=utf8mb4"
SYNC_DATABASE_URL="mysql+pymysql://root:1234@127.0.0.1:3306/hk_db?charset=utf8mb4"


# config.py
class Settings:
    database_url: str
    sync_database_url: str

• 개발 / 운영 DB 설정

  • 개발 환경
    • 변수를 읽어들이는 순서에 따라 먼저 .env.dev에사 읽어들인다.
    • .env.dev에는 값이 설정되어 있으므로 해당 값을 읽는다.
  • 운영 환경
    • 변수를 읽어들이는 순서에 따라 먼저 .env.prod에사 읽어들인다.
    • .env.prod에는 값이 설정되어 있지 않다.
    • 따라서, 다음 순서인 시크릿 파일을 읽어서 값을 읽는다.
  
  
  

• FastAPI DB 연결 설정

  • diriver, username, password를 url 경로에 포함시켜 작성
  • charset=utf8mb4
    • 한글 처리 및 이모지까지 포함하는 설정
  • autocommit=true
    • connection 레벨에서, 즉 MySQL 드랑버 레벨에서 autocommit 처리
    • FastAPI는 SQLAlchemy가 트랜잭션 관리하므로 불필요
  • use_unicode=1
    • MySQL 드라이버가 유니코드 문자열을 제대로 처리하도록 하는 설정
    • 최신 Python MySQL 드라이버에서는 기본값
  • rewriteBatchedStatements=true
    • SQLAlchemy가 자체적으로 배치 처리
  • serverTimezone=Asia/Seoul
    • 대부분 자동 처리하지만 명시적으로 설정 가능
  
  
  

• 기존 Spring Boot 설정

  spring:
  datasource:
      url: jdbc:mysql://127.0.0.1:3306/hk_db?characterEncoding=utf8&autoReconnect=true&serverTimezone=Asia/Seoul&rewriteBatchedStatements=true
      username: root
      password: 1234
      driver-class-name: com.mysql.cj.jdbc.Driver
  

app/core/middleware.py

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
import time
import logging
from app.core.config import settings

logger = logging.getLogger(__name__)

def add_middleware(app: FastAPI):
    app.add_middleware(
        CORSMiddleware,
        allow_origins=settings.cors_origins_list,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"]
    )

    if not settings.is_dev:
        app.add_middleware(
            TrustedHostMiddleware,
            allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
        )

    if settings.is_dev:
        @app.middleware("http")
        async def log_requests(request: Request, call_next):
            start_time = time.time()
            response = await call_next(request)
            process_time = time.time() - start_time

            logger.info(
                f"{request.method} {request.url.path} - "
                f"Status: {response.status_code} - "
                f"Time: {process_time:.4f}s"
            )
            return response

중요 설정

• app.add_middleware( )

  from fastapi import FastAPI, Request
  
  def add_middleware(app: FastAPI):
  
      app.add_middleware(
          TrustedHostMiddleware,
          allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
      )
  

  • add_middleware()
    • FastAPI에 내장된 함수
  
  
  

• 미들웨어 개념

  • Spring Boot의 Filter/Interceptor

    @Component
    public class LoggingFilter implements Filter {
        @Override
        public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) {
            // 요청 전 처리
            chain.doFilter(request, response);  // 다음 필터/컨트롤러로
            // 응답 후 처리
        }
    }
    

    
    
  • FastAPI의 미들웨어

    @app.middleware("http")
    async def logging_middleware(request, call_next):
        # 요청 전 처리
        response = await call_next(request)  # 다음 미들웨어/라우터로
        # 응답 후 처리
        return response
    

    • 미들웨어는 요청/응답을 가로채서 처리하는 중간 계층
    • 인증, 로깅, CORS, 압축 등
  
  
  

• TrustedHostMiddleware

  from fastapi.middleware.trustedhost import TrustedHostMiddleware
  
  def add_middleware(app: FastAPI):
  
      app.add_middleware(
          TrustedHostMiddleware,
          allowed_hosts=["yourdomain.com", "*.yourdomain.com"]
      )
  

  • 악의적인 Host 헤더로 인한 캐시 중독, 리다이렉트 공격을 방지

  • Spring Security에서 Host Header 공격 방지와 비슷한 개념

    @Bean
    public SecurityFilterChain filterChain(HttpSecurity http) {
        http.headers().frameOptions().sameOrigin();
    }
    

  
  
  

• logger

  import logging
  
  logger = logging.getLogger(name)
  

  • **logging.getLogger(__**name**__)**
    • 로거 인스턴스 생성
    • Spring의 LoggerFactory 같은 것
    • __name__
      • 현재 모듈명 ("app.core.middleware")
      • 모듈별로 구분된 로거를 만들어 로그 출처 식별 가능
  
  
  

• @app.middleware("http")

  from fastapi import FastAPI, Request
  import time
  import logging
  
  def add_middleware(app: FastAPI):
  @app.middleware("http")
  async def log_requests(request: Request, call_next):
      start_time = time.time()
      response = await call_next(request)
      process_time = time.time() - start_time
  
      logger.info(
          f"{request.method} {request.url.path} - "
          f"Status: {response.status_code} - "
          f"Time: {process_time:.4f}s"
      )
      return response
  

  • @app.middleware("http")

    • 커스텀 HTTP 미들웨어를 등록하는 데코레이터
    • Spring Boot의 HandlerInterceptor를 함수로 구현한 것
    
    

  • Request

    • Http Request 객체
    • Spring Boot의 HttpServletRequest와 동일한 역할
    
    

  • call_next

    async def middleware(request: Request, call_next):
        # 전처리
        response = await call_next(request)  # 다음 미들웨어/라우터로 전달
        # 후처리
        return response
    

    • call_next는 다음 미들웨어나 실제 라우터 함수를 호출하는 함수

    • 실행 순서

      1. 미들웨어 A 전처리
      2. call_next() → 미들웨어 B 전처리
      3. call_next() → 실제 라우터 함수 실행
      4. 미들웨어 B 후처리
      5. 미들웨어 A 후처리

    • Spring Boot의 FilterChain.doFilter()와 동일한 개념

      public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) {
          // 전처리
          chain.doFilter(request, response);  // 다음 필터/컨트롤러로 전달
          // 후처리
      }
      

app/main.py

from fastapi import FastAPI

from app.core.config import settings
from app.core.middleware import add_middleware

app = FastAPI(
    title="Holiday Keeper API",
    description="공휴일 정보 관리 API",
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc"
)

add_middleware(app)

@app.get("/")
async def root():
    return {"message": "Holiday Keeper API is running!"}

@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "service": "holiday-keeper-api",
        "version": "1.0.0"
    }

@app.get("/test")
async def test():
    return {
        "message": "테스트 성공",
        "data": ["공휴일1", "공휴일2", "공휴일3"]
    }

@app.get("/config")
async def get_config():
    if not settings.is_dev:
        return {"error": "개발 환경에서만 사용 가능"}

    return {
        "project_name": settings.project_name,
        "version": settings.version,
        "environment": settings.environment,
        "server_host": settings.server_host,
        "server_port": settings.server_port,
        "cors_origins": settings.cors_origins_list,
        "log_level": settings.log_level
    }

• middleware.py의 add_middleware(app)를 활용하여 관심사 분리

  
  

• middleware.py 도입 전

  from fastapi import FastAPI
  from fastapi.middleware.cors import CORSMiddleware
  
  app = FastAPI(
      title="Holiday Keeper API",
      description="공휴일 정보 관리 API",
      version="1.0.0",
      docs_url="/docs",
      redoc_url="/redoc"
  )
  
  app.add_middleware(
      CORSMiddleware,
      allow_origins=settings.cors_origins_list,
      allow_credentials=True,
      allow_methods=["*"],
      allow_headers=["*"]
  )
  

app/core/database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import settings
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

sync_engine = create_engine(
    settings.sync_database_url,
    pool_pre_ping=True,
    echo=settings.is_dev
)

async_engine = create_async_engine(
    settings.database_url,
    pool_pre_ping=True,
    echo=settings.is_dev
)

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

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False
)

async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

def get_sync_db():
    sync_db = SessionLocal()
    try:
        yield sync_db
    finally:
        sync_db.close()

DB Engine

from sqlalchemy import create_engine
from sqlalchemy.ext.asyncio import create_async_engine
from app.core.config import settings

sync_engine = create_engine(
    settings.sync_database_url,
    pool_pre_ping=True,
    echo=settings.is_dev
)

async_engine = create_async_engine(
    settings.database_url,
    pool_pre_ping=True,
    echo=settings.is_dev
)

• Engine

  • 데이터베이스 Connection Pool을 관리하는 객체
  • 데이터베이스 연결 생성 및 연결 재사용
  • 엔진은 전역에서 하나만 생성해서 공유하는 것이 권장된다
    • 엔진마다 별도의 연결 풀을 생성하면 낭비이다.
    • 또 엔진 생성에는 비용이 들기 때문에 좋지 않다.
  
  

• create_engine 함수 (동기 엔진)

  • 동기 방식으로 엔진 생성
  • 블로킹 방식
    • 각 쿼리가 완료될 때까지 대기
  
  
  

• create_async_engine 함수 (비동기 엔진)

  • 비동기 방식으로 엔진 생성
  • 논블로킹 방식
    • 쿼리 실행 중에 다른 작업 수행 가능
  
  
  

• engine 파라미터

  • url: str | URL
    • 필수 인자
    • 데이터베에스 연결 URL
  • pool_pre_ping=True
    • connection pool에서 connection을 가져올 때 먼저 ping을 보내 connection 상태 확인
  • echo=settings.is_dev
    • 개발 모드일 때 실행되는 SQL 쿼리를 콘솔에 출력
    • settings.is_dev는 임의로 만든 환경 확인 함수
  • pool_size
    • 연결 풀의 기본 크기
    • default : 5
  • max_overflow
    • 풀이 가득 찰 때 추가로 생성할 수 있는 연결 수
    • default : 10
  • pool_timeout
    • 연결을 얻기 위해 대기할 최대 시간(초)
  • pool_recycle
    • 연결의 최대 생존 시간(초)
    • 여기서 정해진 시간이 지남에 따라 연결을 폐기 후 새로 생성한다.

DB 세션 생성

from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.asyncio import async_sessionmaker

SessionLocal = sessionmaker(
    bind=sync_engine,        # 사용할 엔진
    autocommit=False,        # 자동 커밋 비활성화
    autoflush=False,         # 자동 플러시 비활성화
)

AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,       # 비동기 엔진 바인딩
    autocommit=False,        # 자동 커밋 비활성화
    autoflush=False,         # 자동 플러시 비활성화
    expire_on_commit=False   # 커밋 후 객체 만료 방지
)

• SessionLocal

  • 세션 팩토리 객체
  • SessionLocal()을 호출할 때마다 새로운 세션 인스턴스 생성
  • 각 요청/트랜잭션마다 독립적인 세션 사용
  
  

• sessionmaker

  • 세션을 생성하는 팩토리 함수
  • sessionmaker - (동기)
  • async_sessionmaker - (비동기)
  • 호출할 때마다 새로운 세션 인스턴스 생성
  
  
  

• session 파라미터

  • bind

    • 필수 인자
    • 연결할 엔진 지정

  • autocommit=False

    • ORM 세션 레벨에서의 autocommit 처리
    • 명시적으로 commit()을 호출해야 변경사항이 저장된다
    • 반드시 false로 해야 트랜잭션 관리를 할 수 있다
    • true인 경우, 즉시 DB에 저장되어 롤백 불가능!

  • autoflush=False

    • true : 기본값
    • False로 설정해야 ****쿼리 전에 자동으로 flush하지 않는다
    • 즉, 수동으로 제어하는 것으로 성능상 유리하다.

  • expire_on_commit

    • true : 기본값
    • False로 설정해야 ****커밋 후에도 객체의 속성에 접근 가능
    • expire_on_commit=False가 비동기에서 중요
      • 세션이 닫힌 후에도 해당 객체를 활용하기 위해서

  • class_

    • 사용할 세션 클래스 지정
    • 생략 시 기본값으로 지정된다
      • sessionmaker ⇒ Session
      • async_sessionmaker ⇒ AsyncSession
    • 커스템 세션 클래스의 경우 명시적으로 지정

  • info

    SessionLocal = sessionmaker(
        bind=engine,
        info={
            'app_name': 'my_app',
            'version': '1.0.0',
            'debug_mode': True
        }
    )
    
    # 저장 정보 사용 방법
    session.info['key']
    

    • 추가 세션 메타데이터 딕셔너리

DB 세션 생성

async def get_db():
    async with AsyncSessionLocal() as db:
        yield db

def get_sync_db():
    sync_db = SessionLocal()
    try:
        yield sync_db
    finally:
        sync_db.close()

• async with 문법

  async with AsyncSessionLocal() as session:
      # 세션 사용
      pass  # 자동으로 session.close() 호출됨
  

  • async with는 비동기 컨텍스트 매니저
    • 즉, 세션의 생명주기 관리로 세션 생성/정리를 자동으로 처리
    • 진입 시 : AsyncSessionLocal()로 세션 생성
    • 종료 시 : 자동으로 await session.close() 호출
    • 예외 발생 시 : 자동으로 정리 작업 수행
  • 일반 with와 비슷하지만 __aenter__와 __aexit__ 메서드가 비동기
  
  
  

• try, yield, finally 문법

  async def get_db():
      async with AsyncSessionLocal() as session:
          try:
              yield session          # 세션을 제공
          finally:
              await session.close()  # 정리 작업
  

  • yield
    • 이 함수를 제너레이터로 만든다
    • FastAPI가 의존성 주입할 때 필요한 문법
  • try
    • 세션 사용 중 예외가 발생할 수 있는 구간
  • finally
    • 예외 발생 여부와 상관없이 반드시 실행되는 정리 코드
  
  
  • FastAPI에서는 의존성 주입 시 이 패턴을 사용한다
    1. 요청 시작 시 세션 생성
    2. 요청 처리 중 세션 사용
    3. 요청 완료 후 세션 정리

app/common/models/base.py (Base 모델)

from sqlalchemy import Column, Integer, DateTime, func
from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    pass

class TimestampMixin:
    created_at = Column(DateTime, server_default=func.now(), nullable=False)
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)

class BaseModel(Base, TimestampMixin):
    __abstract__ = True

    id = Column(Integer, primary_key=True)

• DeclarativeBase

  • SQLAlchemy 2.0에서 도입된 새로운 베이스 클래스
    • 메타클래스 자동 설정
      • DeclarativeMeta 메타클래스를 자동으로 적용
      • 복잡한 설정을 자동으로 적용
    • 타입 힌팅 개선
      • Python의 타입 시스템과 더 잘 통합 (IDE에서 자동완성 개선)
      • 타입 검사 지원
    • 레지스트리 관리
      • SQLAlchemy가 모든 테이블 정의를 중앙에서 관리
      • registry : 매핑된 클래스들의 레지스트리
    • 스키마 생성
      • 실제 데이터베이스에 테이블을 만들거나 삭제할 때 사용
        • Base.metadata.create_all(engine)
      • metadata 객체를 통해 테이블 생성/삭제 관리
    
    

  • 이전 방식

    # SQLAlchemy 1.x
    from sqlalchemy.ext.declarative import declarative_base
    
    Base = declarative_base()
    
    # SQLAlchemy 2.0+
    from sqlalchemy.orm import DeclarativeBase
    
    class Base(DeclarativeBase):
        pass
    

  
  
  

• 네이밍 관례

  • Base

    • SQLAlchemy의 루트 베이스 클래스
    • 모든 모델의 최상위 부모
    • 보통 Base 또는 DeclarativeBase로 명명
    
    

  • BaseModel

    • 공통 필드와 로직을 담는 추상 베이스 모델
    • Base를 상속받아 확장한다
    
    

  • TimestampMixin

    • 믹스인 패턴 : 특정 기능을 여러 클래스에 주입
    • 보통 created_at, updated_at 같은 공통 시간 필드 제공
    • 이름에 Mixin 접미사로 목적 명시
  
  
  

• __abstract__ = True

  • __abstract__의 의미

    • 이 클래스는 실제 테이블을 생성하지 않는다고 설정

    • 다른 모델들이 상속받기 위한 공통 속성 정의 역할
    • Spring Boot의 BaseEntity와 유사

    • 예시

      class TimestampMixin:
          created_at = Column(DateTime, server_default=func.now(), nullable=False)
          updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now(), nullable=False)
      
      class BaseModel(Base, TimestampMixin):
          __abstract__ = True  # 테이블 생성 안 함
          id = Column(Integer, primary_key=True)
      
      class Country(BaseModel):  # BaseModel 상속
          __tablename__ = 'countries'  # 실제 테이블 생성
          name = Column(String(100))
      

      • BaseModel ⇒ 테이블 생성 안 됨
      • Country ⇒ countries 테이블 생성
        • id, created_at, updated_at, name 포함
  
  
  

• func

  • SQLAlchemy의 SQL 함수 네임스페이스

  • 날짜/시간

    func.now()                        # 현재 시간
    func.current_date()               # 현재 날짜
    func.current_timestamp()          # 현재 타임스탬프
    func.date_part('year', date_col)  # 날짜 부분 추출
    

    
    

  • 집계 함수

    func.count(User.id)          # COUNT
    func.sum(Order.amount)       # SUM
    func.avg(Product.price)      # AVERAGE
    func.min(User.age)           # MIN
    func.max(User.age)           # MAX
    

    
    

  • 문자열 함수

    func.upper(User.name)        # 대문자 변환
    func.lower(User.email)       # 소문자 변환
    func.length(User.name)       # 문자열 길이
    func.concat(User.first_name, ' ', User.last_name)    # 문자열 연결
    

    
    

  • 수학 함수

    func.abs(column)             # 절댓값
    func.round(column, 2)        # 반올림
    func.floor(column)           # 내림
    func.ceiling(column)         # 올림
    

app/domains/country/models.py (Country 모델)

from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.orm import relationship
from app.common.models.base import BaseModel

class Country(BaseModel):
    __tablename__ = 'countries'

    country_code = Column(String(10), unique=True, nullable=False)
    name = Column(String(100), nullable=False)

    holidays = relationship("Holiday", back_populates="country")

• Column( ) 함수

  Column(type_, *args, **kwargs)
  

  • 데이터 타입

    Column(Integer)           # 정수형
    Column(String(100))       # 최대 100자 문자열
    Column(Text)              # 긴 텍스트
    Column(DateTime)          # 날짜/시간
    Column(Boolean)           # 불린
    Column(Float)             # 실수
    Column(DECIMAL(10, 2))    # 정밀한 소수점
    

    
    

  • 제약 조건

    Column(Integer, primary_key=True)            # 기본키
    Column(String(50), nullable=False)           # NOT NULL
    Column(String(100), unique=True)             # UNIQUE 제약
    Column(Integer, default=0)                   # 기본값
    Column(DateTime, server_default=func.now())  # 서버 기본값
    

    • 서버 기본값이란
      • DB 레벨의 기본값을 말한다.
      • 일반 기본값의 경우, 쿼리를 넘길 때 기본값으로 설정하여 내보내는 것이다.
      • 서버 기본값은 쿼리에서 생략해도 DB 서버가 알아서 NOW()로 입력한다.
    
    

  • 인덱스와 성능

    Column(String(50), index=True)        # 인덱스 생성
    Column(Integer, primary_key=True)     # PK는 자동으로 인덱스
    

    • PK와 UK는 자동으로 인덱스 설정된다
    • FK는 DB마다 다르다
    
    

  • 업데이트 관련

    Column(DateTime, onupdate=func.now())             # 업데이트 시 자동 갱신
    Column(DateTime, server_onupdate=func.now())      # 서버 레벨 업데이트
    

    
    

  • 외래키

    Column(Integer, ForeignKey('users.id'))  # 외래키 설정
    

    • ForeignKey 설정에는 실제 테이블명 사용
    
    

  • 기타 옵션

    Column(String(100), 
        comment='사용자 이름',            # 주석
        info={'label': 'Username'},       # 추가 메타데이터
        quote=True,                       # 컬럼명 따옴표 처리
        autoincrement=True)               # 자동 증가 (PK의 기본값)
    

    • autoincrement=True
      • DB별 자동 증가 전략 선택
      • 다만, PK + Integer의 경우 자동 증가가 자동 적용된다.
    • quote=True

      • 예약어나 특수문자가 포함된 컬럼명인 경우 설정

        ⇒ 따옴표 처리로 가능하게 한다

  
  
  

• relationship( )

  • 기본 관계 설정

    # 현재 코드
    holidays = relationship("Holiday", back_populates="country")
    
    relationship("User")                           # 기본 관계
    relationship("User", backref="orders")         # 역참조 자동 생성
    relationship("User", back_populates="orders")  # 양방향 명시적 설정
    

    • relationship("User", XXX)
      • 대상 클래스는 모델 클래스명을 사용한다.
    
    

  • 외래키 관련

    relationship("User", foreign_keys=[user_id])     # 외래키 명시
    relationship("User", primaryjoin="Order.user_id==User.id")  # 조인 조건 명시
    

    
    

  • 로딩 전략

    relationship("User", lazy="select")      # 기본: 필요시 로딩
    relationship("User", lazy="joined")      # 조인으로 즉시 로딩  
    relationship("User", lazy="subquery")    # 서브쿼리로 로딩
    relationship("User", lazy="dynamic")     # 쿼리 객체 반환
    relationship("User", lazy="selectin")    # IN 절 사용해서 로딩
    

    • Default : lazy="select"
      • 현업에서 가장 많이 사용
      • 기본값 사용 시 생략 가능
      • 테이블 정의는 기본값으로 두고, 쿼리할 때 상황에 맞게 제어
    • 성능 최적화가 필요할 때 ⇒ N+1 문제 해결
      • lazy="joined"
      • lazy="selectin"
    
    

  • 컬렉션 타입

    relationship("Tag", collection_class=set)        # Set으로 관리
    relationship("Tag", collection_class=list)       # List로 관리 (기본값)
    

    • Default : collection_class=list
    
    

  • 삭제 동작

    relationship("Order", 
                cascade="all, delete-orphan",    # 연쇄 삭제
                passive_deletes=True)            # DB 레벨 삭제
    

    • cascade="all”은 여러 cascade 옵션의 조합이다.
      • save-update : 부모 저장 시 자식도 저장
      • merge : 부모 병합 시 자식도 병합
      • refresh : 부모 새로고침 시 자식도 새로고침
      • expunge : 부모 세션에서 제거 시 자식도 제거
      • delete : 부모 삭제 시 자식도 삭제
    • delete-orphan
      • 부모와 연결이 끊어진 자식 자동 삭제

        user = session.query(User).first()
        order = user.orders[0]  
        user.orders.remove(order)  # 관계에서 제거
        session.commit()
        

        • Order 객체가 자동으로 DB에서 삭제됨
    • passive_deletes=True
      • Default : passive_deletes=False
      • False는 삭제 시 SQLAlchemy가 SELECT로 모든 자식들을 찾아 하나씩 DELETE
        • 느리고 메모리 많이 사용
      • True는 삭제 시 DB의 CASCADE가 알아서 관련 자식 삭제
        • 빠르다
    
    

  • 정렬

    relationship("Comment", 
                order_by="Comment.created_at.desc()")  # 정렬 조건
    

    • JPA의 @OrderBy와 동일
    • 데이터를 가져올 때 정렬 조건
    
    

  • 조건부 관계

    class Order(BaseModel):
        # 모든 상품
        product = relationship("Product")
        
        # 활성화된 상품만
        active_product = relationship("Product",
                                    primaryjoin="and_(Order.product_id==Product.id, "
                                            "Product.is_active==True)")
    
    # 사용
    order = session.query(Order).first()
    print(order.product)        # 모든 상품
    print(order.active_product) # 활성화된 상품만 (조건에 맞는 것만)
    

    
    

  • 양방향 관계

    # Country 모델
    class Country(BaseModel):
        holidays = relationship("Holiday", back_populates="country")
    
    # Holiday 모델  
    class Holiday(BaseModel):
        country_id = Column(Integer, ForeignKey('countries.id'))
        country = relationship("Country", back_populates="holidays")
    

    • 직접 FK 설정을 해줘야 실제 연결이 된다
      • 즉, 단방향이든 양방향이든 FK 설정이 완료되어야지 실제 연결된다

app/domains/holiday/models.py (Holiday 모델)

from sqlalchemy import Column, Integer, String, Date, Boolean, UniqueConstraint, Index, ForeignKey
from sqlalchemy.orm import relationship
from app.common.models.base import BaseModel

class Holiday(BaseModel):
    __tablename__ = 'holidays'
    __table_args__ = (
        UniqueConstraint('country_id', 'date', 'name', name='uk_1'),

        Index('idx_year', 'holiday_year'),
        Index('idx_country_year', 'country_id', 'holiday_year'),
    )

    date = Column(Date, nullable=False)
    name = Column(String(200))
    local_name = Column(String(200))
    holiday_year = Column(Integer, nullable=False)
    launch_year = Column(Integer)
    is_global = Column(Boolean, default=False)
    types = Column(String(100))
    counties = Column(String(500))

    country_id = Column(Integer, ForeignKey('countries.id'), nullable=False, index=True)

    country = relationship("Country", back_populates="holidays")

• 1대다 관계 설정

  • Spring Boot의 경우

    • 어노테이션 기반으로 관계 정의
    • mappedBy로 연관관계 주인이 아님을 명시
      • 자동으로 FK는 상대방이 가진다
    
    

  • FastAPI의 경우

    • FK 필드를 통해 관계를 결정
    • FK를 가진 테이블이 연관관계의 주인이 된다
  
  
  

• back_populates vs backref

  • 둘 모두 양방향 설정으로 양쪽에서 모두 연관 대상에 접근할 수 있다

  • backref

    • 한쪽 모델에서만 관계를 정의한다
      • 다른 쪽 모델은 relationship 정의 안 하지만, 자동 생성된다.
    • 간단한 프로젝트에서 주로 사용
    
    

  • back_populates

    • 양쪽 모델 모두에서 관계를 정의한다
    • 명시적으로 설정하여 각 관계가 어떻게 설정되어 있는지 명확하다.
    • 또한 각 관계마다 다른 옵션 설정 가능
    • IDE 지원 : 자동완성과 타입 검사 더 잘됨
  
  
  

• 다대다 관계 설정

  • 단순 다대다 관계 (추가 필드 없는 경우)

    • Spring Boot의 경우

      @Entity
      public class User {
          @ManyToMany
          @JoinTable(
              name = "user_roles",
              joinColumns = @JoinColumn(name = "user_id"),
              inverseJoinColumns = @JoinColumn(name = "role_id")
          )
          private List<Role> roles;
      }
      
      @Entity
      public class Role {
          @ManyToMany(mappedBy = "roles")
          private List<User> users;
      }
      

      
      

    • FastAPI의 경우

      from sqlalchemy import Table
      
      user_role_table = Table(
          'user_roles',
          Base.metadata,
          Column('user_id', Integer, ForeignKey('users.id')),
          Column('role_id', Integer, ForeignKey('roles.id'))
      )
      
      class User(BaseModel):
          __tablename__ = 'users'
          name = Column(String(100))
          
          roles = relationship("Role", secondary=user_role_table, back_populates="users")
      
      class Role(BaseModel):
          __tablename__ = 'roles'
          name = Column(String(50))
          
          users = relationship("User", secondary=user_role_table, back_populates="roles")
      

      • 중간 테이블 정의
    
    
    

  • 복잡한 다대다 관계 (중간 테이블에 추가 필드)

    • Spring Boot의 경우

      @Entity
      public class UserRole extends BaseEntity {
          @ManyToOne(fetch = FetchType.LAZY)
          @JoinColumn(name = "user_id")
          private User user;
      
          @ManyToOne(fetch = FetchType.LAZY)
          @JoinColumn(name = "role_id")
          private Role role;
          
          private LocalDateTime assignedDate;
      }
      
      @Entity
      public class User extends BaseEntity {
          @OneToMany(mappedBy = "user", cascade = CascadeType.ALL, orphanRemoval = true)
          private List<UserRole> userRoles = new ArrayList<>();
      }
      
      @Entity
      public class Role extends BaseEntity {
          @OneToMany(mappedBy = "role", cascade = CascadeType.ALL, orphanRemoval = true)
          private List<UserRole> roleUsers = new ArrayList<>();
      }
      

      
      

    • FastAPI의 경우

      from sqlalchemy import DateTime, func
      
      class UserRole(BaseModel):
          __tablename__ = 'user_roles'
          
          user_id = Column(Integer, ForeignKey('users.id'), primary_key=True)
          role_id = Column(Integer, ForeignKey('roles.id'), primary_key=True)
          assigned_date = Column(DateTime, server_default=func.now())
          
          user = relationship("User", back_populates="user_roles")
          role = relationship("Role", back_populates="role_users")
      
      class User(BaseModel):
          __tablename__ = 'users'
          name = Column(String(100))
          
          user_roles = relationship("UserRole", back_populates="user")
      
      class Role(BaseModel):
          __tablename__ = 'roles' 
          name = Column(String(50))
          
          role_users = relationship("UserRole", back_populates="role")
      

app/common/crud/base.py

from typing import Type, TypeVar, Any, Optional, List
from sqlalchemy import select, func, text
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase

ModelType = TypeVar("ModelType", bound=DeclarativeBase)

class CRUDBase:
    def __init__(self, model: Type[ModelType]):
        self.model = model

    async def get(self, db: AsyncSession, obj_id: Any) -> Optional[ModelType]:
        return await db.get(self.model, obj_id)

    async def get_by_field(self, db: AsyncSession, field_name: str, value: Any) -> Optional[ModelType]:
        stmt = select(self.model).where(getattr(self.model, field_name) == value)
        result = await db.execute(stmt)
        return result.scalars().one_or_none()

    async def get_by_fields(self, db: AsyncSession, **kwargs: Any) -> Optional[ModelType]:
        conditions = [getattr(self.model, key) == value for key, value in kwargs.items()]
        stmt = select(self.model).where(*conditions)
        result = await db.execute(stmt)
        return result.scalars().one_or_none()
    
    async def get_multi(self, db: AsyncSession, *, skip: int = 0, limit: int = 100) -> List[ModelType]:
        stmt = select(self.model).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    async def get_multi_by_fields(self, db: AsyncSession, *, skip: int = 0, limit: int = 100, **kwargs: Any) -> List[ModelType]:
        conditions = [getattr(self.model, key) == value for key, value in kwargs.items()]
        stmt = select(self.model).where(*conditions).offset(skip).limit(limit)
        result = await db.execute(stmt)
        return result.scalars().all()

    async def get_all(self, db: AsyncSession) -> List[ModelType]:
        stmt = select(self.model)
        results = await db.execute(stmt)
        return results.scalars().all()

    async def count(self, db: AsyncSession) -> int:
        stmt = select(func.count()).select_from(self.model)
        result = await db.execute(stmt)
        return result.scalar_one()

    async def create(self, db: AsyncSession, *, obj_in: Any) -> ModelType:
        db_obj = self.model(**obj_in.model_dump())
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return db_obj

    async def update(self, db: AsyncSession, db_obj: ModelType, obj_in: Any) -> ModelType:
        update_data = obj_in.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_obj, key, value)
        db.add(db_obj)
        await db.commit()
        await db.refresh(db_obj)
        return  db_obj

    async def remove(self, db: AsyncSession, *, db_obj: ModelType) -> ModelType:
        await db.delete(db_obj)
        await db.commit()
        return db_obj

초기 설정

• TypeVar

  from typing import Type, TypeVar, Any, Optional, List
  from sqlalchemy.orm import DeclarativeBase
  
  ModelType = TypeVar("ModelType", bound=DeclarativeBase)
  
  class CRUDBase:
      def __init__(self, model: Type[ModelType]):
          self.model = model
  

  • TypeVar란

    • Python의 타입 힌팅 시스템에서 제네릭 타입 변수를 정의하는 문법
    
    

  • "ModelType"

    • TypeVar의 이름을 문자열로 지정
    • 관례적으로 변수명과 동일하게 설정
    
    

  • bound=DeclarativeBase

    • TypeVar가 가질 수 있는 타입의 upper bound을 제한
    • 즉, 이 경우 DeclarativeBase를 상속받은 클래스만 올 수 있다
      • SQLAlchemy ORM 모델들은 모두 DeclarativeBase를 상속받는다
    
    
    

  • 그 외 파라미터

    # 제약 없는 TypeVar
    T = TypeVar('T')
    
    # 특정 타입들로만 제한
    NumberType = TypeVar('NumberType', int, float, complex)
    
    # 공변성/반공변성 설정
    T_co = TypeVar('T_co', covariant=True)              # 공변
    T_contra = TypeVar('T_contra', contravariant=True)  # 반공변
    
    # bound와 constraints 조합
    UserType = TypeVar('UserType', bound=BaseModel, int, str)
    

    
    
    

  • 현업 명명 사례

    # 모델/엔티티 관련
    ModelType = TypeVar('ModelType', bound=BaseModel)
    EntityT = TypeVar('EntityT', bound=Entity)
    DomainT = TypeVar('DomainT')
    
    # 서비스/리포지토리 패턴
    ServiceT = TypeVar('ServiceT', bound=BaseService)
    RepoT = TypeVar('RepoT', bound=BaseRepository)
    
    # HTTP 관련
    RequestT = TypeVar('RequestT', bound=BaseRequest)
    ResponseT = TypeVar('ResponseT', bound=BaseResponse)
    
    # 일반적인 명명
    T = TypeVar('T')  # 가장 일반적
    K = TypeVar('K')  # Key 타입
    V = TypeVar('V')  # Value 타입
    

  
  
  
  

• __init__

  • __init__ 함수는 Python의 생성자 메서드

  • __init__ 함수

    • Python의 인스턴스 초기화 메서드 (생성자)
    • 객체가 생성될 때 자동으로 호출
    • 인스턴스의 초기 상태를 설정하는 역할
    
    
    

  • self

    • 인스턴스 자신을 가리키는 참조
    • 인스턴스 메서드의 첫 번째 파라미터로 자동 전달
    • 정적 메서드가 아닌 인스턴스 메서드임을 의미
    
    
    

  • model: Type[ModelType]

    • Type[ModelType]
      • 제네릭 타입 변수 ModelType의 타입을 받는다는 타입 힌트
    • 인스턴스가 아닌 클래스 자체를 받는다

단건 / 다건 조회

• SQLAlchemy ORM

  • 클래스는 테이블의 메타데이터를 담고 있는 설계도 역할을 한다.
  • 객체를 통해 메서드를 호출하지만 내부에서는 클래스 자체만을 사용
  
  
  
  

• 단건 조회

  async def get(self, db: AsyncSession, obj_id: Any) -> Optional[ModelType]:
      return await db.get(self.model, obj_id)
  
  async def get_by_field(self, db: AsyncSession, field_name: str, value: Any) -> Optional[ModelType]:
      stmt = select(self.model).where(getattr(self.model, field_name) == value)
      result = await db.execute(stmt)
      return result.scalars().one_or_none()
  
  async def get_by_fields(self, db: AsyncSession, **kwargs: Any) -> Optional[ModelType]:
      conditions = [getattr(self.model, key) == value for key, value in kwargs.items()]
      stmt = select(self.model).where(*conditions)
      result = await db.execute(stmt)
      return result.scalars().one_or_none()
  

  • SQLAlchemy의 get( ) 함수

    • Primary Key로 단일 객체 조회하는 SQLAlchemy 메서드
    • db.get(self.model, obj_id)
    • 첫 번째 인자 : 조회할 모델 클래스
    • 두 번째 인자 : Primary Key 값
    • 반환 : 해당 객체 또는 None
    
    
    

  • SQLAlchemy의 select( ) 함수

    stmt = select(User)  # SELECT * FROM user
    stmt = select(User.name, User.email)  # SELECT name, email FROM user
    

    • 쿼리 빌더 함수로, SQL SELECT 문을 생성하는 SQLAlchemy 메서드
    
    
    

  • SQLAlchemy의 where( ) 함수

    # 단일 조건
    stmt = select(User).where(User.id == 1)
    
    # 여러 조건 (AND)
    stmt = select(User).where(User.active == True, User.age > 18)
    
    # 또는 (OR)
    stmt = select(User).where(User.active == True).where(User.age > 18)
    

    • SELECT 문에 WHERE 조건을 추가하는 SQLAlchemy 메서드
    
    
    

  • getattr( ) 함수

    class User:
        name = "John"
        age = 25
    
    # 일반적인 속성 접근
    user.name  # "John"
    
    # getattr을 사용한 동적 속성 접근
    getattr(user, "name")  # "John"
    getattr(user, "salary", 0)  # 기본값 0 (속성이 없을 때)
    

    • 객체의 속성을 동적으로 가져오는 Python 내장 함수
    • 해당 속성이 없을 때 기본값을 지정해서 가져올 수 있다.
    
    
    

  • **kwargs 문법

    def get_by_fields(self, db: AsyncSession, **kwargs: Any):
        # kwargs = {"name": "John", "age": 25, "active": True}
        print(kwargs)  # {'name': 'John', 'age': 25, 'active': True}
        
        # 딕셔너리 순회
        for key, value in kwargs.items():
            print(f"{key}: {value}")
    
    # 사용 예시
    repo.get_by_fields(db, name="John", age=25, active=True)
    

    • 키워드 인자들을 딕셔너리로 받는 Python 문법
    • 딕셔너리 순회를 통해 많이 사용한다.
      • dict 클래스의 메서드인 items() 등 활용
    • 2개 이상의 필드로 조건부 조회하는 경우 정확한 숫자를 몰라도 설정 가능
  
  
  
  

• SQLAlchemy의 execute( ) 함수

  # 비동기 실행
  result = await db.execute(stmt)
  
  # 동기 실행
  result = db.execute(stmt)
  

  • 데이터베이스에 쿼리를 실행하는 SQLAlchemy 메서드

  • execute() 결과의 기본 구조

    # 직접 fetchall() 사용
    rows = result.fetchall()
    print(type(rows[0]))  # <class 'sqlalchemy.engine.Row'>
    
    # Row 객체 접근
    for row in rows:
        # 방법 1: 인덱스로 접근
        print(row[0])  # 첫 번째 컬럼
        
        # 방법 2: 속성명으로 접근
        print(row.id)     # User.id
        print(row.name)   # User.name
        
        # 방법 3: 딕셔너리처럼 접근
        print(row['id'])
        print(row['name'])
        
        # Row를 딕셔너리로 변환
        user_dict = row._asdict()
        print(user_dict)  # {'id': 1, 'name': 'John', 'email': 'john@example.com'}
    

  • Row 객체는 ORM 모델 객체의 메서드를 사용할 수 없다.
  
  
  
  

• SQLAlchemy의 scalars( ) 함수

  # ORM 객체
  result = await db.execute(select(User))
  users = result.scalars().all()
  user_obj = users[0]
  print(type(user_obj))
  user_obj.some_method()  # User 모델의 메서드 사용 가능
  

  • Result 객체에서 스칼라 값들만 추출하는 SQLAlchemy 메서드

  • scalars( ) 대신 사용할 수 있는 메서드들

    • mappings() - 딕셔너리 형태로 결과 반환

      result = await db.execute(select(User))
      user_dicts = result.mappings().all()
      # [{'id': 1, 'name': 'John', 'email': 'john@example.com'}, ...]
      
      # API 응답에 바로 사용 가능
      return JSONResponse(content=user_dicts)
      

      
      

    • 튜플 결과

      result = await db.execute(select(User.id, User.name))
      tuples = result.all()  # [(1, 'John'), (2, 'Jane'), ...]
      
      # 또는 scalars().all()로 첫 번째 컬럼만
      result = await db.execute(select(User.name))
      names = result.scalars().all()  # ['John', 'Jane', ...]
      

    
    
    

  • 뒤에 올 수 있는 메서드

    # 모든 결과 가져오기
    .all()          # 리스트로 모든 결과 반환
    .fetchall()     # all()과 동일
    
    # 단일 결과 가져오기
    .first()        # 첫 번째 결과 또는 None
    .one()          # 정확히 하나의 결과 (없거나 여러 개면 예외)
    .one_or_none()  # 하나의 결과 또는 None (여러 개면 예외)
    
    # 고유값들만
    .unique()       # 중복 제거
    
    # 파티셔닝
    .partitions(size)  # 지정된 크기로 결과를 나눔
    

    • all( ) 메서드의 특징
      • 데이터가 없는 경우 빈 리스트 [] 반환 (None이 아님)
      • 항상 리스트 타입 보장
  
  
  
  
  

• SQLAlchemy의 scalar( ) 함수

  • 쿼리 결과의 첫 번째 행, 첫 번째 컬럼의 값을 추출

    result.scalar()              # 첫 번째 행의 첫 번째 컬럼 (None 가능)
    result.scalar_one()          # 정확히 1개 값 (0개나 2개 이상이면 예외)
    result.scalar_one_or_none()  # 1개 값 또는 None (2개 이상이면 예외)
    

    
    
    

  • 사용 대상

    • COUNT, SUM, MAX, MIN 등 집계 함수 결과
    • 단일 값을 반환하는 쿼리
    • 존재 여부 확인 (EXISTS)
    
    
    

  • scalar( )와 scalars( )의 차이점

    • scalar() - 단일 값만 반환
    • scalars().first() - ORM 객체 하나
    • scalars().all() - ORM 객체 리스트
    • 즉, 일반 상세 조회 등에서는 scalar( )를 쓰지 않는다. 객체가 필요하기 때문에
    
    
    

  • 사용 예시

    # 1. COUNT - 가장 많이 사용
    async def get_user_count(db: AsyncSession) -> int:
        result = await db.execute(select(func.count(User.id)))
        return result.scalar()  # 정수 반환 (예: 150)
    
    # 2. 조건부 COUNT
    async def get_active_user_count(db: AsyncSession) -> int:
        result = await db.execute(
            select(func.count(User.id)).where(User.active == True)
        )
        return result.scalar()
    
    # 3. MAX, MIN 값 조회
    async def get_latest_user_id(db: AsyncSession) -> Optional[int]:
        result = await db.execute(select(func.max(User.id)))
        return result.scalar()  # 최대 ID 값
    
    # 4. SUM 계산
    async def get_total_order_amount(db: AsyncSession, user_id: int) -> float:
        result = await db.execute(
            select(func.sum(Order.amount)).where(Order.user_id == user_id)
        )
        return result.scalar() or 0.0
    
    # 5. EXISTS 확인 (매우 자주 사용)
    async def email_exists(db: AsyncSession, email: str) -> bool:
        result = await db.execute(
            select(exists().where(User.email == email))
        )
        return result.scalar()  # True/False
    
    # 6. 단일 컬럼 값 하나만 가져오기
    async def get_user_name(db: AsyncSession, user_id: int) -> Optional[str]:
        result = await db.execute(
            select(User.name).where(User.id == user_id)
        )
        return result.scalar()  # 이름 문자열 또는 None
    

  
  
  
  

• 다건 조회

  async def get_multi(self, db: AsyncSession, *, skip: int = 0, limit: int = 100) -> List[ModelType]:
      stmt = select(self.model).offset(skip).limit(limit)
      result = await db.execute(stmt)
      return result.scalars().all()
  
  async def get_multi_by_fields(self, db: AsyncSession, *, skip: int = 0, limit: int = 100, **kwargs: Any) -> List[ModelType]:
      conditions = [getattr(self.model, key) == value for key, value in kwargs.items()]
      stmt = select(self.model).where(*conditions).offset(skip).limit(limit)
      result = await db.execute(stmt)
      return result.scalars().all()
  
  async def get_all(self, db: AsyncSession) -> List[ModelType]:
          stmt = select(self.model)
          results = await db.execute(stmt)
          return results.scalars().all()
  

  • offset( )과 limit( ) 함수

    stmt = select(User).offset(10).limit(5)
    # 생성되는 SQL: SELECT * FROM user LIMIT 5 OFFSET 10
    # 10개를 건너뛰고 5개만 가져와라
    

    • SQLAlchemy의 페이징 처리를 위한 SQL 함수들
    
    
    

  • * 키워드 전용 인자 문법

    def func(a, b, *, c, d):
        pass
    
    # 올바른 호출
    func(1, 2, c=3, d=4)  # ✅
    
    # 잘못된 호출  
    func(1, 2, 3, 4)      # ❌ TypeError: 위치 인자로 c, d 전달 불가
    

    • * 이후의 매개변수들은 반드시 키워드로 전달해야 한다

      
      

    • 키워드 인자와 위치 인자

      • 위치 인자의 경우, 순서에 따라 매개변수에 값이 할당된다.

        greet("김철수", 25, "서울")     # ✅
        greet(25, "서울", "김철수")     # ❌ or 의도와 다른 결과
        

      • 키워드 인자의 경우, 매개변수 이름을 명시하여 할당한다.

        greet(name="김철수", age=25, city="서울")  # ✅
        greet(city="서울", name="김철수", age=25)  # ✅
        

      
      

    • 이렇게 키워드 인자로 사용하는 이유

      users = await crud.get_multi(db, skip=20, limit=10)
      

      • 무엇을 건너뛰고 몇 개 가져올지 명확하게 명시하기 위해

집계 함수

async def count(self, db: AsyncSession) -> int:
    stmt = select(func.count()).select_from(self.model)
    result = await db.execute(stmt)
    return result.scalar_one()

async def count_by_fields(self, db: AsyncSession, **kwargs: Any) -> int:
    conditions = [getattr(self.model, key) == value for key, value in kwargs.items()]
    stmt = select(func.count()).select_from(self.model).where(*conditions)
    result = await db.execute(stmt)
    return result.scalar_one()

• 사용 예시

  total_users = await user_crud.count(db)  # 전체 사용자 수
  active_users = await user_crud.count_by_fields(db, active=True)  # 활성 사용자 수
  

  
  
  

• scalar_one( )

  • scalar_one()

    • 오직 단 하나의 값만 나와야 한다
    • scalar_one()을 사용하는 이유는 COUNT 함수는 항상 숫자를 반환하기 때문
    • 조건에 맞는 데이터가 없는 경우 null이 아닌 0을 반환
    
    
    

  • 다른 집계 함수

    # MAX, MIN, SUM, AVG - NULL 가능
    SELECT MAX(age) FROM users;           -- 데이터 없으면 NULL
    SELECT SUM(salary) FROM users;        -- 데이터 없으면 NULL  
    SELECT AVG(age) FROM users;           -- 데이터 없으면 NULL
    

    • 그러므로, 다른 집계함수에서는 scalar()를 쓰거나 scalar_one_or_none() 사용
    • scalar()는 결과가 여러 개여도 첫번째 것만 가져온다
    • scalar_one_or_none()는 결과가 여러 개라면 예외를 발생시킨다
  
  
  
  

• SQLAlchemy의 select_from( ) 함수

  select(User)  
  # SQL: SELECT * FROM user
  
  select(func.count()).select_from(User)
  # SQL: SELECT COUNT(*) FROM user
  

  • FROM 절을 명시적으로 지정하는 함수

    • 일반적인 select() 함수는 FROM이 자동 추론된다.

    • 다만, 집계 함수를 쓰는 경우 FROM 절이 불분명한 경우가 있다

    • 컬럼을 명시하는 경우 ⇒ select_from 불필요

      stmt = select(func.count(User.id))
      

    • 제네릭 타입이므로 컬럼을 명시하기 힘들다
    
    
    

  • 서브 쿼리로 FROM 절을 사용하는 경우도 select_from 필요

    stmt = select(func.count()).select_from(select(User).subquery())
    # SQL: SELECT COUNT(*) FROM (SELECT * FROM user) AS subquery
    

생성 / 수정 / 삭제

async def create(self, db: AsyncSession, *, obj_in: Any) -> ModelType:
    db_obj = self.model(**obj_in.model_dump())
    db.add(db_obj)
    await db.commit()
    await db.refresh(db_obj)
    return db_obj

async def update(self, db: AsyncSession, db_obj: ModelType, obj_in: Any) -> ModelType:
    update_data = obj_in.model_dump(exclude_unset=True)
    for key, value in update_data.items():
        setattr(db_obj, key, value)
    db.add(db_obj)
    await db.commit()
    await db.refresh(db_obj)
    return  db_obj

async def remove(self, db: AsyncSession, *, db_obj: ModelType) -> ModelType:
    await db.delete(db_obj)
    await db.commit()
    return db_obj

• obj_in

  • Pydantic Schemas가 들어오기 때문에 Any로 설정
  • Spring Boot에서 DTO를 받아 전환 후 DB 저장하는 패턴과 유사
  
  
  
  

• model_dump( ) 함수

  • Pydantic BaseModel 클래스의 메서드

  • Pydantic 객체를 딕셔너리로 변환한다.

    # Pydantic 객체 생성
    user_data = UserCreate(name="김철수", email="kim@email.com", age=30)
    
    # model_dump() 호출 - 딕셔너리로 변환
    print(user_data.model_dump())
    # 출력: {'name': '김철수', 'email': 'kim@email.com', 'age': 30, 'active': True}
    

    
    
    

  • model_dump( )의 매개변수들

    • 기본 사용

      user_data.model_dump()
      # {'name': '김철수', 'email': 'kim@email.com', 'age': 30, 'active': True}
      

      
      
    • exclude_unset=True

      user_data.model_dump(exclude_unset=True)
      # {'name': '김철수', 'email': 'kim@email.com', 'age': 30}  # active는 기본값이므로 제외
      

      • 설정되지 않은 필드 제외
      • 기본값 : False
      • 일부 필드만 업데이트하는 경우 중요하다
      
      
    • include

      user_data.model_dump(include={'name', 'email'})
      # {'name': '김철수', 'email': 'kim@email.com'}
      

      • 특정 필드만 포함
      
      
    • exclude

      user_data.model_dump(exclude={'active'})
      # {'name': '김철수', 'email': 'kim@email.com', 'age': 30}
      

      • 특정 필드 제외
      
      
    • by_alias=True

      user_data.model_dump(by_alias=True)
      

      • 필드 별칭 사용
      
      
    • exclude_none=True

      user_data.model_dump(exclude_none=True)
      

      • None 값 제외
  
  
  
  

• refresh( ) 함수

  • 데이터베이스와 객체를 동기화한다

  • 데이터 삽입 과정

    async def create(self, db: AsyncSession, *, obj_in: Any) -> ModelType:
        # 1. Pydantic → SQLAlchemy 객체 생성
        db_obj = self.model(**obj_in.model_dump())
        
        print(db_obj.id)  # None (아직 DB에 저장 전)
        
        # 2. 세션에 추가
        db.add(db_obj)
        print(db_obj.id)  # 여전히 None
        
        # 3. DB에 커밋 (실제 저장)
        await db.commit()
        print(db_obj.id)  # 여전히 None!
        
        # 4. 객체 새로고침 (DB에서 최신 데이터 가져옴)
        await db.refresh(db_obj)
        print(db_obj.id)  # 1 (DB에서 자동 생성된 ID)
        
        return db_obj
    

    1. Pydantic의 model_dump() 함수로 딕셔너리 전환
    2. 딕셔너리를 self.model(해당 모델)의 init 함수로 넘겨 모델 객체 생성
    3. db.add()로 DB 세션에 저장
    4. db.commit()을 통해 실제 DB에 저장
    5. db.refresh()를 통해 해당 객체의 정보를 DB에서 갱신
    6. Database의 종류가 무엇이든, SQLAlchemy는 DB 저장 후 ID를 가져온다
  
  
  
  

• setattr( ) 함수

  class User:
      def __init__(self):
          self.name = ""
          self.age = 0
  
  user = User()
  
  # 일반적인 속성 설정
  user.name = "김철수"
  user.age = 25
  
  # setattr()을 사용한 동적 속성 설정
  setattr(user, "name", "김철수")  # user.name = "김철수"와 같음
  setattr(user, "age", 25)         # user.age = 25와 같음
  

  • getattr() 함수는 동적으로 객체의 속성의 값을 가져오는 것이다
  • setattr()은 동적으로 객체의 속성의 값을 설정하는 것이다