20250417-2

이상우·2025년 4월 17일

네, .env 파일 예시와 이를 사용하여 FastAPI에서 데이터베이스(비동기 PostgreSQL 기준)에 연결하는 코드 예시를 보여드리겠습니다.

1. .env 파일 예시

프로젝트 루트 디렉토리 (예: your_project_root/)에 .env 파일을 생성하고 아래 내용을 필요에 맞게 수정하여 추가합니다. 주의: .env 파일은 민감한 정보를 포함하므로 .gitignore에 추가하여 Git 저장소에 커밋되지 않도록 해야 합니다.

# .env

# PostgreSQL 데이터베이스 연결 정보 (비동기 드라이버 asyncpg 사용 예시)
# 형식: postgresql+asyncpg://사용자명:비밀번호@호스트주소:포트번호/데이터베이스명
DATABASE_URL=postgresql+asyncpg://your_db_user:your_db_password@localhost:5432/your_db_name

# 다른 DB 예시 (필요시 주석 해제 및 수정)
# MySQL (aiomysql 드라이버):
# DATABASE_URL=mysql+aiomysql://your_db_user:your_db_password@localhost:3306/your_db_name
# SQLite (비동기):
# DATABASE_URL=sqlite+aiosqlite:///./your_database_file.db

# FastAPI 애플리케이션 보안을 위한 시크릿 키 (JWT 토큰 등에 사용)
# 터미널에서 openssl rand -hex 32 등으로 생성 가능
SECRET_KEY=your_very_secret_key_here_change_this

# API 버전 접두사 (선택 사항)
API_V1_STR=/api/v1

# IP 제한을 위한 허용 IP 목록 (쉼표로 구분)
# 실제 운영 환경에 맞게 수정하세요.
ALLOWED_IPS=127.0.0.1,192.168.1.100,YOUR_OFFICE_PUBLIC_IP

# CORS 설정 (React 개발 서버 및 실제 도메인)
# 쉼표로 구분된 문자열 또는 JSON 형식의 리스트 문자열로 저장 가능
# 예시 1: 쉼표 구분 문자열
BACKEND_CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000,https://your-admin-domain.com
# 예시 2: JSON 형식 리스트 문자열 (파싱 필요)
# BACKEND_CORS_ORIGINS='["http://localhost:3000", "http://127.0.0.1:3000", "https://your-admin-domain.com"]'

# 프로젝트 이름 (선택 사항)
PROJECT_NAME=My FastAPI Admin Project

# 서버 포트 (선택 사항, uvicorn 실행 시 우선 적용)
# PORT=8000

2. FastAPI에서 DB 연결 및 설정 로드 예시 (app/core/config.py, app/db/session.py, app/main.py)

필요한 라이브러리 설치:

pip install "fastapi[all]" sqlalchemy[asyncio] asyncpg python-dotenv pydantic pydantic-settings
# 또는 MySQL 사용 시: pip install aiomysql
# 또는 SQLite 사용 시: pip install aiosqlite

a) 설정 로드 (app/core/config.py)

Pydantic의 BaseSettings를 사용하여 .env 파일을 자동으로 읽고 타입 검증까지 수행합니다.

# app/core/config.py
import os
from typing import List, Union, Set
from pydantic import AnyHttpUrl, validator
from pydantic_settings import BaseSettings # pydantic v2 이상
# from pydantic import BaseSettings # pydantic v1의 경우

# .env 파일 로드 (프로젝트 루트에 .env 파일이 있다고 가정)
# load_dotenv() # python-dotenv 라이브러리가 설치되어 있다면 자동으로 로드될 수도 있음
# 명시적으로 로드하려면:
# from dotenv import load_dotenv
# load_dotenv()

class Settings(BaseSettings):
    PROJECT_NAME: str = "FastAPI Project"
    API_V1_STR: str = "/api/v1"

    # 데이터베이스 설정
    DATABASE_URL: str

    # 보안 설정
    SECRET_KEY: str

    # CORS 설정
    # 문자열 리스트 또는 AnyHttpUrl 리스트로 받을 수 있음
    BACKEND_CORS_ORIGINS: Union[str, List[AnyHttpUrl]]

    # IP 제한 설정
    ALLOWED_IPS: str # 문자열로 읽어온 후 파싱하거나, validator 사용

    # BACKEND_CORS_ORIGINS를 문자열 리스트로 파싱하는 validator 예시
    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[List[str], str]:
        if isinstance(v, str) and not v.startswith("["):
            # 쉼표로 구분된 문자열이라면 리스트로 변환
            return [i.strip() for i in v.split(",")]
        elif isinstance(v, (list, str)):
            # 이미 리스트거나 JSON 형식 문자열이면 그대로 반환 (Pydantic이 처리)
            return v
        raise ValueError(v)

    # ALLOWED_IPS를 Set[str]으로 파싱하는 validator 예시
    @validator("ALLOWED_IPS", pre=True)
    def assemble_allowed_ips(cls, v: str) -> Set[str]:
        if isinstance(v, str):
             # 쉼표로 구분된 문자열을 Set으로 변환 (중복 제거 및 빠른 검색)
            return {ip.strip() for ip in v.split(",")}
        raise ValueError(v)

    class Config:
        # .env 파일 경로 명시 (기본값은 현재 작업 디렉토리의 .env)
        # case_sensitive = True # 환경 변수 이름 대소문자 구분 여부
        env_file = ".env"
        env_file_encoding = 'utf-8'

# 설정 객체 인스턴스 생성 (다른 모듈에서 import하여 사용)
settings = Settings()

# --- 사용 예시 ---
# 다른 파일에서:
# from app.core.config import settings
# print(settings.DATABASE_URL)
# print(settings.ALLOWED_IPS) # Set 형태로 파싱됨: {'127.0.0.1', '192.168.1.100', 'YOUR_OFFICE_PUBLIC_IP'}
# print(settings.BACKEND_CORS_ORIGINS) # List 형태로 파싱됨

b) 데이터베이스 세션 관리 (app/db/session.py)

SQLAlchemy를 사용하여 비동기 데이터베이스 엔진과 세션 메이커를 설정하고, FastAPI 의존성 함수(get_db)를 만듭니다.

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import sessionmaker # 동기식 예시 참고용

from app.core.config import settings # 설정 로드

# 비동기 엔진 생성
# echo=True는 실행되는 SQL 쿼리를 로깅합니다 (개발 시 유용)
async_engine = create_async_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True, # 커넥션 풀에서 연결 가져오기 전에 테스트 핑 실행 여부
    echo=False # 운영 환경에서는 False 권장
)

# 비동기 세션 메이커 생성
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession, # 비동기 세션 사용 명시
    autocommit=False, # 자동 커밋 비활성화 (명시적 커밋/롤백 권장)
    autoflush=False, # 자동 flush 비활성화
    expire_on_commit=False # 커밋 후 객체 만료 방지 (FastAPI에서 유용)
)

# FastAPI 의존성 함수 (비동기)
# 각 API 요청마다 독립적인 DB 세션을 제공하고, 요청 완료 후 세션을 닫습니다.
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            # yield 이후의 코드는 요청 처리가 끝난 후에 실행됩니다.
            # (성공 시) 특별히 할 작업이 없다면 비워둡니다.
            # 기본적으로는 예외가 발생하지 않으면 자동으로 커밋되지 않습니다.
            # 라우터 함수 내에서 await session.commit() 필요
        except Exception as e:
            # 라우터에서 예외 발생 시 롤백 (선택적이지만 권장)
            await session.rollback()
            raise e # 예외를 다시 발생시켜 FastAPI가 처리하도록 함
        # finally 블록은 필요 없습니다. `async with`가 세션 close를 자동으로 처리합니다.

# --- 동기식 참고 (사용하지 않을 경우 제거) ---
# sync_engine = create_engine(settings.DATABASE_URL, pool_pre_ping=True)
# SyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=sync_engine)
# def get_sync_db():
#     db = SyncSessionLocal()
#     try:
#         yield db
#     finally:
#         db.close()

c) FastAPI 앱 설정 (app/main.py)

FastAPI 앱 인스턴스를 생성하고, CORS 미들웨어, IP 체크 미들웨어(이전 답변 참고), API 라우터 등을 설정합니다.

# app/main.py
from fastapi import FastAPI, Depends, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware

# 설정, DB 세션, 라우터 등 import
from app.core.config import settings
from app.db.session import get_db, AsyncSession # 비동기 세션 의존성
# from app.api.routers.admin import users as admin_users_router # 예시 라우터
# from app.middlewares.ip_check import IPCheckMiddleware # IP 체크 미들웨어 (별도 파일 구현 가정)

# --- IP 체크 미들웨어 예시 (main.py에 직접 정의) ---
class IPCheckMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        client_host = request.client.host
        # 설정 파일에서 ALLOWED_IPS (Set) 가져오기
        allowed_ips = settings.ALLOWED_IPS

        # '/admin' 경로 등 특정 경로에만 적용하거나 전체 적용
        # if request.url.path.startswith("/admin"): # 예시: /admin 경로
        if client_host not in allowed_ips:
            print(f"Forbidden IP: {client_host}") # 로그
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"IP address {client_host} is not allowed.",
            )
        response = await call_next(request)
        return response

# FastAPI 앱 인스턴스 생성
app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json" # API 문서 경로
)

# CORS 미들웨어 설정
# settings.BACKEND_CORS_ORIGINS는 config.py에서 파싱된 리스트
if settings.BACKEND_CORS_ORIGINS:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS] if isinstance(settings.BACKEND_CORS_ORIGINS, list) else [],
        allow_origin_regex=settings.BACKEND_CORS_ORIGINS if isinstance(settings.BACKEND_CORS_ORIGINS, str) else None, # 정규식 사용 시
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

# IP 체크 미डल웨어 등록
# CORS 미들웨어 뒤, 라우터 앞에 등록하는 것이 일반적
app.add_middleware(IPCheckMiddleware)

# API 라우터 등록 (예시)
# app.include_router(admin_users_router, prefix=settings.API_V1_STR + "/admin/users", tags=["admin-users"])

# 기본 경로 (테스트용)
@app.get("/")
async def read_root():
    return {"message": f"Welcome to {settings.PROJECT_NAME}"}

# DB 연결 테스트용 엔드포인트 (예시)
@app.get("/db-test")
async def test_db_connection(db: AsyncSession = Depends(get_db)):
    try:
        # 간단한 쿼리 실행 (실제 테이블/데이터 없이 연결만 확인)
        # 예: SELECT 1;
        result = await db.execute(text("SELECT 1"))
        if result.scalar() == 1:
            return {"message": "Database connection successful!"}
        else:
             raise HTTPException(status_code=500, detail="Database connection test failed.")
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Database connection error: {e}")

# SQLAlchemy의 text 함수 import 필요
from sqlalchemy.sql import text

이제 uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 명령어로 FastAPI 서버를 실행하면, .env 파일의 설정을 읽어 데이터베이스에 연결하고, 설정된 IP 주소 및 CORS 정책을 적용합니다.

알겠습니다. FastAPI는 Python 프레임워크이므로 Java의 JDBC(Java Database Connectivity)를 직접 사용하지 않습니다. Python에서는 데이터베이스 연결을 위해 DB-API라는 표준 인터페이스를 따르는 드라이버를 사용하며, 보통 SQLAlchemy와 같은 ORM(Object-Relational Mapper) 라이브러리와 함께 사용합니다.

MySQL에 비동기적으로 연결하기 위해 FastAPI 및 SQLAlchemy와 함께 사용할 수 있는 대표적인 Python 드라이버는 aiomysql 또는 asyncmy입니다. 여기서는 aiomysql을 기준으로 예시를 보여드리겠습니다.

1. .env 파일 예시 (MySQL용)

프로젝트 루트에 .env 파일을 만들고 아래 내용을 MySQL 환경에 맞게 수정합니다.

# .env

# MySQL 데이터베이스 연결 정보 (비동기 드라이버 aiomysql 사용 예시)
# 형식: mysql+aiomysql://사용자명:비밀번호@호스트주소:포트번호/데이터베이스명?charset=utf8mb4
DATABASE_URL=mysql+aiomysql://your_mysql_user:your_mysql_password@localhost:3306/your_mysql_db?charset=utf8mb4

# --- 다른 설정들은 이전 예시와 동일하게 유지 ---

# FastAPI 애플리케이션 보안을 위한 시크릿 키
SECRET_KEY=your_very_secret_key_here_change_this

# API 버전 접두사 (선택 사항)
API_V1_STR=/api/v1

# IP 제한을 위한 허용 IP 목록 (쉼표로 구분)
ALLOWED_IPS=127.0.0.1,192.168.1.100,YOUR_OFFICE_PUBLIC_IP

# CORS 설정 (React 개발 서버 및 실제 도메인)
BACKEND_CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000,https://your-admin-domain.com

# 프로젝트 이름 (선택 사항)
PROJECT_NAME="My FastAPI Admin (MySQL)"

주의:

DATABASE_URL의 시작 부분이 mysql+aiomysql://로 변경되었습니다.
MySQL 사용자명, 비밀번호, 호스트, 포트(기본값 3306), 데이터베이스명을 실제 환경에 맞게 수정해야 합니다.
?charset=utf8mb4를 추가하여 UTF-8 인코딩을 명시하는 것이 좋습니다 (이모지 등 지원).

2. FastAPI에서 DB 연결 및 설정 로드 예시 (MySQL + aiomysql)

필요한 라이브러리 설치:

# aiomysql 드라이버와 SQLAlchemy, FastAPI 등 설치
pip install "fastapi[all]" sqlalchemy[asyncio] aiomysql python-dotenv pydantic pydantic-settings

a) 설정 로드 (app/core/config.py)

이 파일은 이전 PostgreSQL 예시와 거의 동일하게 유지됩니다. Pydantic BaseSettings는 .env 파일에서 DATABASE_URL 문자열을 읽어오기만 할 뿐, 그 내용이 PostgreSQL용인지 MySQL용인지는 신경 쓰지 않습니다.

# app/core/config.py
# (이전 PostgreSQL 예시와 동일한 코드를 사용합니다)
import os
from typing import List, Union, Set
from pydantic import AnyHttpUrl, validator
from pydantic_settings import BaseSettings # pydantic v2 이상

class Settings(BaseSettings):
    PROJECT_NAME: str = "FastAPI Project"
    API_V1_STR: str = "/api/v1"

    # 데이터베이스 설정 (문자열로 읽어옴)
    DATABASE_URL: str

    # 보안 설정
    SECRET_KEY: str

    # CORS 설정
    BACKEND_CORS_ORIGINS: Union[str, List[AnyHttpUrl]]

    # IP 제한 설정
    ALLOWED_IPS: str

    @validator("BACKEND_CORS_ORIGINS", pre=True)
    def assemble_cors_origins(cls, v: Union[str, List[str]]) -> Union[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)

    @validator("ALLOWED_IPS", pre=True)
    def assemble_allowed_ips(cls, v: str) -> Set[str]:
        if isinstance(v, str):
            return {ip.strip() for ip in v.split(",")}
        raise ValueError(v)

    class Config:
        env_file = ".env"
        env_file_encoding = 'utf-8'

settings = Settings()

# 사용 예시 확인
# from app.core.config import settings
# print(settings.DATABASE_URL) # .env 파일에 정의된 MySQL URL 출력됨
# print(settings.ALLOWED_IPS)

b) 데이터베이스 세션 관리 (app/db/session.py)

이 파일도 SQLAlchemy의 추상화 덕분에 PostgreSQL 예시와 거의 동일합니다. create_async_engine 함수는 DATABASE_URL의 접두사(mysql+aiomysql://)를 보고 자동으로 적절한 드라이버를 사용합니다.

# app/db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.core.config import settings # 설정 로드

# 비동기 엔진 생성 (MySQL URL 사용)
async_engine = create_async_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,
    echo=False # SQL 쿼리 로깅 여부
)

# 비동기 세션 메이커 생성 (이 부분은 DB 종류와 무관)
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False
)

# FastAPI 의존성 함수 (비동기) - 이 부분도 DB 종류와 무관
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            # 성공 시 명시적 커밋은 라우터에서 수행
        except Exception as e:
            await session.rollback() # 예외 발생 시 롤백
            raise e
        # `async with`가 자동 close 처리

c) FastAPI 앱 설정 (app/main.py)

이 파일도 DB 연결 설정 자체와는 직접적인 관련이 없으므로, PostgreSQL 예시와 동일하게 유지됩니다. get_db 의존성을 통해 얻는 session 객체는 SQLAlchemy AsyncSession 타입이므로, 사용하는 방식은 같습니다.

# app/main.py
# (이전 PostgreSQL 예시와 동일한 코드를 사용합니다)
from fastapi import FastAPI, Depends, HTTPException, Request, status
from fastapi.middleware.cors import CORSMiddleware
from starlette.middleware.base import BaseHTTPMiddleware
from sqlalchemy.sql import text # DB 테스트용

from app.core.config import settings
from app.db.session import get_db, AsyncSession # get_db 가져오기
# from app.api.routers... # 라우터 가져오기
# from app.middlewares... # 미들웨어 가져오기

# IP 체크 미들웨어 정의 (이전 예시와 동일)
class IPCheckMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        client_host = request.client.host
        allowed_ips = settings.ALLOWED_IPS # 설정에서 로드
        # if request.url.path.startswith("/admin"): # 경로 기반 체크
        if client_host not in allowed_ips:
            print(f"Forbidden IP: {client_host}")
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"IP address {client_host} is not allowed.",
            )
        response = await call_next(request)
        return response

app = FastAPI(
    title=settings.PROJECT_NAME,
    openapi_url=f"{settings.API_V1_STR}/openapi.json"
)

# CORS 미들웨어 설정 (이전 예시와 동일)
if settings.BACKEND_CORS_ORIGINS:
    app.add_middleware(
        CORSMiddleware,
        allow_origins=[str(origin) for origin in settings.BACKEND_CORS_ORIGINS] if isinstance(settings.BACKEND_CORS_ORIGINS, list) else [],
        allow_origin_regex=settings.BACKEND_CORS_ORIGINS if isinstance(settings.BACKEND_CORS_ORIGINS, str) else None,
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

# IP 체크 미들웨어 등록 (이전 예시와 동일)
app.add_middleware(IPCheckMiddleware)

# 라우터 등록 (예시)
# app.include_router(...)

@app.get("/")
async def read_root():
    return {"message": f"Welcome to {settings.PROJECT_NAME}"}

# DB 연결 테스트용 엔드포인트 (DB 종류와 무관하게 동작)
@app.get("/db-test")
async def test_db_connection(db: AsyncSession = Depends(get_db)):
    try:
        # SQLAlchemy는 DB 종류에 맞는 SQL 실행
        result = await db.execute(text("SELECT 1"))
        if result.scalar() == 1:
            return {"message": "MySQL Database connection successful!"}
        else:
             raise HTTPException(status_code=500, detail="Database connection test failed.")
    except Exception as e:
        # 연결 실패 시 구체적인 오류 메시지 확인 가능
        raise HTTPException(status_code=500, detail=f"Database connection error: {e}")

결론:

SQLAlchemy와 같은 ORM을 사용하면, 데이터베이스 종류가 변경되더라도 코드 수정이 최소화된다는 큰 장점이 있습니다. 설정 파일(.env)의 DATABASE_URL을 변경하고, 해당 데이터베이스에 맞는 Python 드라이버(aiomysql 등)를 설치하는 것만으로 대부분의 코드를 재사용할 수 있습니다. JDBC는 Java 생태계의 기술이며, Python에서는 DB-API 드라이버와 SQLAlchemy를 사용하는 것이 표준적인 방법입니다.

네, autoreconnect=True와 useCursorFetch=True 옵션을 데이터베이스 연결 시 적용하는 방법을 설명해 드리겠습니다.

하지만 중요한 점이 있습니다:

autoreconnect: 이 옵션은 특정 Python DB 드라이버(예: mysql-connector-python의 동기 버전)에서 지원하는 기능이지만, aiomysql (비동기 드라이버)에서는 직접적인 autoreconnect 연결 파라미터가 공식적으로 문서화되어 있지 않습니다. 비동기 환경에서는 연결 풀링(SQLAlchemy의 pool_recycle, pool_pre_ping)을 통해 연결 상태를 관리하는 것이 더 일반적입니다. aiomysql 자체가 내부적으로 재연결 로직을 일부 가질 수는 있으나, 명시적인 autoreconnect=True 파라미터는 아닙니다. 대신 SQLAlchemy의 pool_recycle 옵션을 사용하여 주기적으로 연결을 재활용(끊고 다시 맺음)하는 것이 끊어진 연결 문제를 해결하는 표준적인 방법입니다.
useCursorFetch=True: 이 옵션은 주로 Java의 JDBC 드라이버에서 사용되는 파라미터로, 서버 측 커서(server-side cursor)를 사용하여 대량의 데이터를 한 번에 가져오지 않고 스트리밍 방식으로 처리할 때 사용됩니다. Python SQLAlchemy + aiomysql 환경에서는 이와 동일한 이름의 연결 파라미터가 없습니다. SQLAlchemy에서 서버 측 커서와 유사한 효과(결과 스트리밍)를 얻으려면, 쿼리 실행 시 stream_results=True 또는 yield_per 같은 실행 옵션(execution option)을 사용해야 합니다. 이는 연결 시점이 아닌, 특정 쿼리를 실행할 때 지정하는 옵션입니다.

결론적으로, 요청하신 두 옵션을 모든 DB 연결 시 자동으로 적용하는 것은 Python/SQLAlchemy/aiomysql 환경에서는 직접적으로 불가능하거나 권장되지 않습니다.

대신, 유사한 목적을 달성하는 SQLAlchemy의 표준적인 방법을 적용해 보겠습니다.

autoreconnect -> SQLAlchemy의 pool_recycle 사용 (주기적으로 연결 재생성)
useCursorFetch -> 해당 기능은 쿼리 실행 시 옵션으로 지정 (전역 설정 불가)

수정된 app/db/session.py 예시 (MySQL + aiomysql 기준):

# app/db/session.py
import os
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from app.core.config import settings # 설정 로드

# 연결 풀 설정을 위한 파라미터
# pool_recycle: 지정된 시간(초)마다 연결을 재활용합니다.
# MySQL의 기본 wait_timeout 값(보통 8시간 = 28800초)보다 짧게 설정하는 것이 일반적입니다.
# 예: 1시간 = 3600초
MYSQL_POOL_RECYCLE = 3600

# Driver-specific connection arguments (필요한 경우)
# aiomysql에서 특정 옵션을 connection string 대신 여기에 넣어야 할 때 사용
connect_args = {}

# # 만약 aiomysql 드라이버에 직접 전달해야 하는 다른 옵션이 있다면 여기에 추가
# if settings.DATABASE_URL.startswith("mysql+aiomysql"):
#     # 예시: connect_args["ssl_mode"] = "VERIFY_IDENTITY" 등
#     pass

# 비동기 엔진 생성
async_engine = create_async_engine(
    settings.DATABASE_URL,
    pool_pre_ping=True,       # 커넥션 풀에서 연결 가져오기 전 테스트 핑 (권장)
    pool_recycle=MYSQL_POOL_RECYCLE, # 지정된 시간마다 연결 재활용 (autoreconnect 대안)
    echo=False,               # SQL 쿼리 로깅 여부
    connect_args=connect_args # 드라이버에 직접 전달할 추가 인자
)

# 비동기 세션 메이커 생성 (이 부분은 이전과 동일)
AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    autocommit=False,
    autoflush=False,
    expire_on_commit=False
)

# FastAPI 의존성 함수 (비동기) - 이전과 동일
async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
        except Exception as e:
            await session.rollback()
            raise e

설명:

pool_recycle=MYSQL_POOL_RECYCLE: 이 옵션이 autoreconnect의 가장 가까운 대안입니다. 설정된 시간(여기서는 3600초 = 1시간)이 지난 유휴(idle) 연결은 자동으로 끊고 다시 연결합니다. 이는 MySQL 서버가 장시간 유휴 상태인 연결을 먼저 끊어버리는(wait_timeout) 문제로 인한 "MySQL server has gone away" 오류를 예방하는 데 효과적입니다. MySQL의 wait_timeout 설정을 확인하고 그보다 짧은 값으로 설정하는 것이 좋습니다.
pool_pre_ping=True: 커넥션 풀에서 연결을 가져올 때마다 간단한 테스트 쿼리(예: SELECT 1)를 실행하여 연결이 유효한지 확인합니다. 유효하지 않으면 새로운 연결을 맺습니다. 이는 pool_recycle 주기 사이에 연결이 끊어졌을 경우에도 대처할 수 있게 해줍니다.

useCursorFetch 대체: 앞서 설명했듯이, 이는 연결 옵션이 아닙니다. 대량의 데이터를 처리해야 하는 특정 쿼리에서만 다음과 같이 실행 옵션을 사용해야 합니다.

# 예시: 특정 라우터에서 대량 데이터 조회 시
from sqlalchemy.future import select
from sqlalchemy import Result # 비동기 결과 처리

async def get_large_data(db: AsyncSession = Depends(get_db)):
    stmt = select(MyLargeTableModel) # MyLargeTableModel은 실제 모델 이름으로 대체

    # stream_results=True 또는 execution_options(stream_results=True) 사용
    # stream_results=True 는 SQLAlchemy 2.0 스타일
    # execution_options는 이전 버전 호환
    result: Result = await db.stream(stmt) # SQLAlchemy 2.0+ stream() 메소드 사용

    # 또는 execution_options 사용
    # result = await db.execute(stmt.execution_options(stream_results=True))

    # 결과를 하나씩 처리 (메모리에 모두 로드하지 않음)
    async for row in result.scalars():
        # 각 row 처리 로직
        process_row(row)

    # 또는 청크 단위 처리
    # async for partition in result.partitions(1000): # 1000개씩 묶어서 처리
    #     process_chunk(partition)

    return {"message": "Large data processed"}

요약:

Python/SQLAlchemy 환경에서는 JDBC의 autoreconnect나 useCursorFetch와 정확히 동일한 이름의 연결 시점 옵션은 없습니다.
autoreconnect 의 목적(연결 유지/복구)은 SQLAlchemy의 pool_recycle 과 pool_pre_ping 옵션을 통해 달성하는 것이 표준입니다.
useCursorFetch 의 목적(서버 측 커서/결과 스트리밍)은 쿼리 실행 시 stream_results=True 또는 yield_per 같은 실행 옵션을 통해 구현합니다. 전역적으로 적용할 수 없습니다.

따라서 위 app/db/session.py 코드처럼 pool_recycle과 pool_pre_ping을 설정하는 것이 요청하신 내용에 대한 가장 적절한 Python/SQLAlchemy 방식의 구현입니다.

이상우

다음 포스트

20250417-2

20260820

0개의 댓글