[FastAPI] Relational Database

 

🐰 FastAPI 공식문서를 보면서 개인적으로 정리한 글 입니다.

 

먼저 DB를 준비한다. 이 포스트에서는 MariaDB를 사용한다.

 

DB의 내용은 다음과 같다.

• DB명 : test
• Table 명 : User

idUser	username	password	nickname
1	admin1	SHA2(admin1,256)	Admin1
2	admin2	SHA2(admin2,256)	Admin2
3	pupba12	SHA2(qwer1234,256)	Pupba

 

 

ORM(Object-Relational Mapping)

FastAPI는 ORM을 SQLAlchemy를 사용해 구현한다.

pip install sqlalchemy

 

만든 ORM 모듈의 구조는 다음과 같다.

 

1. SQLAlchemy - database.py

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
import json
path = "./sql_app/secret.json"
se = json.loads(open(path).read())
R_SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{se.get('S_USER')}:{se.get('S_PW')}@{se.get('HOST')}:{se.get('PORT')}/{se.get('DB')}?charset=utf8"
CUD_SQLALCHEMY_DATABASE_URL = f"mysql+pymysql://{se.get('M_USER')}:{se.get('M_PW')}@{se.get('HOST')}:{se.get('PORT')}/{se.get('DB')}?charset=utf8"


r_engine = create_engine(R_SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
cud_engine = create_engine(CUD_SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
r_SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=r_engine)
cud_SessionLocal = sessionmaker(
    autocommit=False, autoflush=False, bind=cud_engine)

Base = declarative_base()

 

create_engine(url, **kwargs)

• 데이터베이스와의 연결을 설정하는 엔진을 생성
• url : 데이터베이스의 연결정보를 담고 있는 URL
• **kwargs : 추가적인 설정을 위한 매개변수(풀링, 문자인코딩 등)

 

sessionmaker(class_=None,bind=None,autoflush=True,autocommit=False,expire_on_commit=True,**kwargs)

• 세션 객체를 생성하기 위한 팩토리 함수, 세션은 SQLAlchemy에서 DB와의 트랜잭션 단위로 작업하기 위한 핵심 객체
• class_ : 생성될 세션 클래스의 부모 클래스
• bind : 연결 엔진을 지정하여 세션과 데이터베이스를 바인딩
• autoflush : 세션에서 변경된 객체를 자동으로 데이터베이스에 반영할지 여부를 설정
• autocommit : 세션이 자동으로 커밋할지 여부를 설정
• expire_on_commit : 커밋 후에 세션에서 로드된 객체를 만료(expire)할지 여부를 설정
• **kwargs : 추가적인 설정을 위한 매개변수

 

declarative_base(cls=object,name='Base',metadata=None,mapper=None,class_registry=None,metaclass=DeclarativeMeta)

• ORM을 사용하여 데이터베이스 테이블과 파이썬 클래스를 매핑하기 위한 기본 클래스 생성
• cls : 생성될 기본 클래스의 부모 클래스
• name : 생성될 기본 클래스의 이름
• metadata : 메타데이터 객체를 지정하여 테이블 생성, 변경 등의 작업을 수행할 수 있음
• mapper : 사용자 정의 매퍼를 지정하여 클래스와 테이블 간의 매핑을 정의할 수 있음
• class_registry : 클래스 레지스트리 객체를 지정하여 클래스들을 등록할 수 있음
• metaclass : 생성된 클래스의 메타클래스를 지정

 

 

2. 데이터베이스 모델 생성 - models.py

데이터베이스에서 SQLAlchemy 모델 생성을 위해 database.py에서 생성한 Base 클래스를 가져와 상속하여 모델을 정의한다.

# Table type
from sqlalchemy import Column, Integer, VARCHAR
# 테이블 관계를 위한 import
# from sqlalchemy.orm import relationship
from sql_app.database import Base


class User(Base):
    """
    User Model
        - id(INT,PRI,UNIQ,NOT NULL): id
        - username(String,UNIQ,NOT NULL) : login id
        - password(String,UNIQ,NOT NULL) : login pw, save to SAH2 HashData
        - nickname(String,UNIQ,NOT NULL) : nickname
    Description : User Table
    """
    # SQLAlchemy에게 각 모델의 데이터베이스에서 사용할 테이블 이름
    __tablename__ = "User"

    idUser = Column(Integer, primary_key=True, unique=True,
                    nullable=False, autoincrement=True)
    username = Column(VARCHAR(45), unique=True, nullable=False)
    password = Column(VARCHAR(64), nullable=False)
    nickname = Column(VARCHAR(45), unique=True, nullable=False)

    """
    - 테이블간 관계 설정 -
    relationship(target, **kwargs)->target 모델을 조작할 수 있는 객체
    - back_populates :  관계의 반대쪽에서 해당 관계를 역참조할 수 있는 속성을 지정합니다. 이를 통해 양방향 관계를 설정할 수 있습니다.
    - uselist : 관계가 리스트(list) 형태인지 여부를 나타냅니다. 기본값은 True이며, False로 설정하면 관계가 스칼라(scalar) 형태로 정의됩니다.
    - lazy : 관계의 로딩 전략을 지정합니다. select로 설정하면 필요한 경우에만 관련된 객체를 로딩하고, joined로 설정하면 쿼리 시 조인을 사용하여 한 번에 관련된 객체를 로딩합니다.
    - cascade : 관계가 변경되었을 때 연관된 객체에 어떤 동작을 전파할지 지정합니다. 예를 들어, delete를 설정하면 부모 객체가 삭제될 때 연관된 자식 객체도 함께 삭제됩니다.
    """

 

 

3. Pydantic 모델  만들기 - schemas.py

읽기용 모델을 위해 pydantic 모델을 만든다.

from typing import List, Union
from pydantic import BaseModel

# Only One User


class UserIn(BaseModel):
    """
    UserIn
        - id : int or None
        - passwd : str
        - username : str
        - nickname : str
    """
    idUser: Union[int, None] = None
    username: str
    password: str
    nickname: str

    class Config:
        from_attributes = True


class UserOut(BaseModel):
    """
    UserOut
        - id : int
        - username : str
        - nickname : str
    """
    idUser: int
    username: str
    nickname: str

    class Config:
        from_attributes = True
# Users


class UsersOut(BaseModel):
    """
    UsersOut
        - users : List[UserOut]
    """
    users: List[UserOut]

    class Config:
        from_attributes = True

 

여기서 orm_mode는 데이터베이스에서 가져온 데이터의 필드와 Pydantic 모델의 필드를 일치시켜, 유효성 검사를 비활성화하고 ORM과 함께 사용할 대 데이터의 변환 및 직렬화 작업을 보다 편리하게 수행할 수 있다.

 

 

4. CRUD 유틸리티 - crud.py

CRUD는 데이터베이스의 데이터를 "생성, 읽기, 업데이트, 삭제" 하는 것을 의미한다.

from sqlalchemy.orm import Session
from sql_app import models, schemas
from hashlib import sha256

# Create


def createUser(db: Session, user: schemas.UserIn):
    sha2PW = sha256(user.password.encode()).hexdigest()
    db_user = models.User(
        idUser=user.idUser,
        username=user.username,
        password=sha2PW,
        nickname=user.nickname
    )
    db.add(db_user)  # 인스턴스를 데이터베이스 세션에 추가
    db.commit()  # 데이터베이스 커밋
    db.refresh(db_user)  # 생성된 ID와 같은 데이터베이스의 새 데이터를 포함하도록 인스턴스 생성
    return db_user

# Read


def getUser(db: Session, username: str):
    return db.query(models.User).filter(models.User.username == username).first()


def getUsers(db: Session):
    return db.query(models.User).offset(0).limit(100).all()

# Update


def updateUser(db: Session, target: models.User, updateUser: schemas.UserIn):
    # To Dict
    updatedData = updateUser.model_dump(exclude_unset=True)
    if updatedData['username'] != target.username:
        target.username = updatedData['username']
    if updatedData['password'] != target.password:
        target.password = updatedData['password']
    if updatedData['nickname'] != target.nickname:
        target.nicknmae = updatedData['nickname']

    db.commit()
    db.refresh(target)
    return target
# Delete


def deleteUser(db: Session, target: models.User):
    db.delete(target)
    db.commit()
    return target

 

 

5. 사용 - main.py

from typing import Union, Dict
from sql_app import crud, models, schemas
from sql_app.database import r_SessionLocal, cud_SessionLocal, r_engine, cud_engine, R_SQLALCHEMY_DATABASE_URL, CUD_SQLALCHEMY_DATABASE_URL
from sqlalchemy.orm import Session
from fastapi import FastAPI, Depends, HTTPException, Form
from fastapi.middleware.cors import CORSMiddleware

models.Base.metadata.create_all(bind=r_engine)
models.Base.metadata.create_all(bind=cud_engine)

app = FastAPI()

# Dependency
app.add_middleware(
    CORSMiddleware,
    allow_origins=[R_SQLALCHEMY_DATABASE_URL, CUD_SQLALCHEMY_DATABASE_URL],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)


def get_rdb():
    db = r_SessionLocal()
    try:
        yield db
    finally:
        db.close()


def get_cuddb():
    db = cud_SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post('/user/', response_model=schemas.UserOut, tags=['GET'])
async def get_user(username: str, db: Session = Depends(get_rdb)):
    result = crud.getUser(db, username=username)
    if result is None:
        raise HTTPException(status_code=404, detail="User not found")
    return result


@app.post('/users/', response_model=schemas.UsersOut, tags=['GET'])
async def get_users(db: Session = Depends(get_rdb)):
    results = crud.getUsers(db)
    return {"users": results}

forms = {
    'target': {'min_length': 5, 'max_length': 15, 'title': 'Target', 'description': "Change Target ID"},
    'username': {'min_length': 5, 'max_length': 15, 'title': 'ID', 'description': "Input ID"},
    'password': {'min_length': 8, 'max_length': 15, 'title': 'Password', 'description': "Input Password"},
    'nickname': {'min_length': 2, 'max_length': 12, 'title': 'NickName', 'description': "Input Nickname"}
}


@app.post('/signin/', response_model=Union[schemas.UserOut, Dict[str, str]], tags=['Create'])
async def signin(
        username: str = Form(..., **forms['username']),
        password: str = Form(..., **forms['password']),
        nickname: str = Form(..., **forms['nickname']),
        rdb: Session = Depends(get_rdb),
        cuddb: Session = Depends(get_cuddb)):
    result = crud.getUser(rdb, username=username)
    if (result.username if result is not None else None) == username:
        return {'msg': "ID is Already...."}
    else:
        user = schemas.UserIn(
            username=username, password=password, nickname=nickname)
        result = crud.createUser(cuddb, user)
        return result


@app.post('/update/', response_model=Union[schemas.UserOut, Dict[str, str]], tags=['Update'])
async def update(
        target: str = Form(..., **forms['target']),
        username: str = Form(..., **forms['username']),
        password: str = Form(..., **forms['password']),
        nickname: str = Form(..., **forms['nickname']),
        rdb: Session = Depends(get_rdb),
        cuddb: Session = Depends(get_cuddb)):
    result = crud.getUser(rdb, username=target)
    if result == None:
        return {'msg': "No Signed Sign In Please....."}
    else:
        # cud 세선에 result 추가
        rdb.expunge(result)  # 세션 분리
        cuddb.add(result)  # 세션 추가
        user = schemas.UserIn(
            username=username, password=password, nickname=nickname)
        result = crud.updateUser(cuddb, result, user)
        return result


@app.post('/delete/', response_model=Union[schemas.UserOut, Dict[str, str]], tags=['Delete'])
async def deleteUser(username: str, rdb: Session = Depends(get_rdb), cuddb: Session = Depends(get_cuddb)):
    result = crud.getUser(rdb, username=username)
    if result == None:
        return {'msg': "No User..."}
    else:
        # cud 세선에 result 추가
        rdb.expunge(result)  # 세션 분리
        cuddb.add(result)  # 세션 추가
        result = crud.deleteUser(cuddb, result)
        return result

 

 

 

jsonable_encoder

FastAPI는 JSON 호환 가능 데이터만 수신하는 데이터베이스와의 통신을 위해 jsonable_encoder() 함수를 제공한다.

ex. datetime 객체는 JSON과 호환되지 않으므로 데이터베이스에 저장불가능, str로 변환되어야 함.

 

jsonable_encoder()에 값을 매개변수로 전달하면 자동으로 바꿔준다.

from fastapi.encoders import jsonable_encoder
from datetime import datetime

from pydantic import BaseModel


class Model(BaseModel):
    title: str
    timestamp: datetime
    desc: str


d = datetime.now()
m = Model(title='HI', timestamp=d, desc="test....")
print(f'{d}-> type: {type(d)}')
print(f'{m}-> type: {type(m)}')
print()
print(f'{jsonable_encoder(d)}-> type: {type(jsonable_encoder(d))}')
print(f'{jsonable_encoder(m)}-> type: {type(jsonable_encoder(m))}')

 

Model은 dict으로 datetime은 str로 바꿔준다.