Fast API - ORM

Fast API&MySQL

 

from sqlalchemy.orm import Session
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

# 1. 데이터베이스 엔진 생성
DATABASE_URL = "mysql+pymysql://user:password@localhost:3306/db_name"
engine = create_engine(DATABASE_URL)

# 2. 베이스 클래스 생성
Base = declarative_base()

# 3. 모델 정의 (DB 테이블 매핑)
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# 4. 테이블 생성
Base.metadata.create_all(bind=engine)

 

데이터베이스 설정

• DATABASE_URL: MySQL 연결 정보 (사용자: user, 비밀번호: password, DB: db_name)
  • mysql+pymysql: SQLAlchemy에게 MySQL을 사용하고, 통신에 pymysql 라이브러리 사용함을 알림
• engine: 데이터베이스 엔진 생성
• Base: ORM 모델 베이스 클래스, declarative_base()=모든 모델이 상속받아야하는 기본 클래스

 

테이블 자동 생성

• 정의된 모델 기반으로 DB 테이블 자동 생성
• SQLAlchemy의 모델 클래스를 실제 데이터베이스의 테이블로 변환
• 테이블이 이미 존재 시 무시함. 개발에 사용

 

세션

from sqlalchemy.orm import sessionmaker

# 세션 팩토리 생성
SessionLocal = sessionmaker(
    autocommit=False,  # 자동 커밋 비활성화
    autoflush=False,    # 자동 플러시 비활성화
    bind=engine
)

# 의존성 주입을 위한 세션 관리 함수
def get_db():
    db = SessionLocal()
    try:
        yield db  # 요청 처리 동안 세션 사용
    finally:
        db.close()  # 요청 종료 시 세션 반환

• 데이터베이스 세션 관리
• 데이터베이스에 쿼리를 보내려면 "세션"이 필요하다.
• 세션은 일종의 연결 통로로, 쿼리를 보내고 결과를 받고, 변경사항을 커밋하거나 롤백하는 역할
• SQLAlchemy에서 데이터베이스와의 상호작용을 관리하는 객체. 따라서, CRUD를 위해서는 세션이 필요하다.
• 2가지 방안이 존재한다. (Depends, sessionMaker)

 

Depends

 

FastAPI의 Depends 기능과 함께 사용하면, 각 요청마다 세션을 자동으로 열고, 요청이 끝나면 자동으로 닫아준다.

의존성 주입을 통한 DB 세션 관리

요청 완료 시 세션 자동 종료

db = Session(bind=engine): 새 세션(연결)을 생성
try:
    yield db: 이 세션을 FastAPI 엔드포인트 함수에 전달
finally: 
    db.close(): 요청이 끝나면(정상/에러 모두) 세션을 닫아 리소스 누수를 방지

 

파이썬에서 yield는 함수 내에서 사용되면 그 함수를 "제너레이터 함수"로 만들어주는 키워드

일반적으로 함수에서 return을 사용하면 값을 한 번만 반환하고 함수가 끝나지만, yield를 사용하면 값을 하나 반환한 뒤 함수 실행이 멈추고, 다음에 다시 호출되면 그 다음 줄부터 실행을 이어간다.

즉, yield는 값을 "하나씩" 반환하면서 함수 실행을 잠시 멈췄다가, 다음에 또 실행되면 멈췄던 부분부터 이어서 실행할 수 있게 해준다.

이렇게 만들어진 함수(제너레이터 함수)는 반복자(iterator)처럼 동작하며, 메모리를 아끼면서 순차적으로 데이터를 처리할 수 있다.

FastAPI에서 get_db() 함수에 yield를 쓰는 이유는 엔드포인트 함수가 DB 세션을 사용할 수 있게 전달해주고, 요청 처리가 끝나면 finally 블록에서 세션을 안전하게 닫을 수 있게 해주기 때문이다.

즉, yield를 기준으로 앞쪽 코드는 "세션을 열어서 엔드포인트에 전달"하는 역할, yield 뒤쪽 finally 코드는 "요청이 끝나면 세션을 닫는 정리 작업"을 담당

 

Pydantic 모델

# MemoCreate: 메모 생성용 필수 필드. 즉, 사용자가 메모를 만들 때 두 필드를 반드시 보내야 하며, 타입이 다르거나 누락되면 자동으로 에러를 반환
# MemoUpdate: 메모 수정용 선택적 필드. 즉, 사용자가 둘 중 하나만 보내도 되고, 둘 다 보내도 되고, 아무것도 안 보내면 아무 변화도 없다. -> PATCH

class MemoCreate(BaseModel):
    title: str
    content: str

class MemoUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None

 

Pydantic의 BaseModel을 상속받아 만든 클래스는 "데이터 유효성 검사"와 "자동 문서화"를 담당한다.

FastAPI는 API 요청을 받을 때, body에 담긴 JSON 데이터를 이 BaseModel로 변환하고, 타입과 필수/선택 여부를 검사한다.

왜 필요한가?

• 보안: 잘못된 데이터가 DB에 들어가는 것을 막음
• 자동 문서화: API 문서(Swagger 등)에 자동 반영
• 코드 간결화: 데이터 검증 로직을 일일이 작성할 필요 없다.

 

의존성 주입 설정

클라이언트로부터 전달받은 데이터를 UserSchema모델로 받도록 설정, Depends(get_db)는 매개변수로 정의한 의존성 함수 get_db함수명을 넣음. 해당 함수의 반환값이 Session 객체이므로, 타입은 Session으로 작성가능

• 데이터베이스 객체 생성: new_user=User(): User 모델 인스턴스 생성
• 데이터베이스 세션에 객체 추가: db.add(new_user): 생성된 User인스턴스를 현재의 데이터베이스 세션에 추가
• 데이터베이스에 커밋: db.commit(): 테이블 객체 내용을 변경하면 db.commit()호출해서 데이터베이스 세션에 대한 변경사항을 DB에 반영
• 객체정보 갱신: db.refresh(new_user): db.refresh()를 호출하여 데이터베이스에 커밋된 새로운 user인스턴스의 정보를 최신상태로 갱신
  • commit은 파이썬에서 변경한 내용을 DB에 저장하는 동작. 즉, 파이썬 → DB 
  • refresh는 DB에 이미 저장된(커밋된) 값을 DB에서 파이썬 객체로 다시 읽어오는 동작. 즉, DB → 파이썬 

 

이를통해 각 요청의 라이프사이클 동안 DB 세션을 효율적으로 관리. Pydantic모델 사용한 데이터 유효성 검증과 SQLAlchemey모델을 통한 데이터베이스 작업 수행 가능

Depends통해 주입된 get_db()함수는 데이터베이스 연결 유지하고, 모든 처리 완료 후 해당 연결 종료

 

CRUD

from fastapi import Depends

# Create
@app.post("/users")
def create_user(user_data: UserSchema, db: Session = Depends(get_db)):
    new_user = User(name=user_data.name, email=user_data.email)
    db.add(new_user)
    db.commit()  # 변경사항 저장
    db.refresh(new_user)  # DB에서 새로고침
    return new_user

# Read
@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    return user

# Update
@app.put("/users/{user_id}")
def update_user(user_id: int, update_data: UserUpdate, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    user.name = update_data.name
    db.commit()
    db.refresh(user)
    return user

# Delete
@app.delete("/users/{user_id}")
def delete_user(user_id: int, db: Session = Depends(get_db)):
    user = db.query(User).filter(User.id == user_id).first()
    db.delete(user)
    db.commit()
    return {"message": "User deleted"}

 

 

1. 메모 생성 (POST)

 

동작: 새 메모 생성 → DB 저장 → 생성된 객체 반환

반환값: 생성된 메모의 ID, 제목, 내용

 

db: Depends(get_db)로 DB 세션이 자동 주입

역할: 데이터베이스와 통신할 수 있는 세션 객체

어떻게 채워지나?

Depends(get_db)는 FastAPI의 의존성 주입 기능

FastAPI가 get_db() 함수를 실행해서, 반환된 db 세션을 이 파라미터에 자동으로 넣어준다.

 

db.commit(): 실제로 DB에 저장

db.refresh(new_memo): DB에서 새로 생성된 메모의 정보를 다시 읽어옴(예: id)

'''

 

5. Pydantic 스키마

역할과 필요성
Pydantic의 BaseModel을 상속받아 만든 클래스는 "데이터 유효성 검사"와 "자동 문서화"를 담당합니다.
FastAPI는 API 요청을 받을 때, body에 담긴 JSON 데이터를 이 BaseModel로 변환하고, 타입과 필수/선택 여부를 검사합니다.
왜 필요한가?
보안: 잘못된 데이터가 DB에 들어가는 것을 막음
자동 문서화: API 문서(Swagger 등)에 자동 반영
코드 간결화: 데이터 검증 로직을 일일이 작성할 필요 없음

MemoCreate: 메모 생성용 필수 필드. 즉, 사용자가 메모를 만들 때 두 필드를 반드시 보내야 하며, 타입이 다르거나 누락되면 자동으로 에러를 반환
MemoUpdate: 메모 수정용 선택적 필드. 즉, 사용자가 둘 중 하나만 보내도 되고, 둘 다 보내도 되고, 아무것도 안 보내면 아무 변화도 없다. -> PATCH

'''