python

FastAPI, CRUD API 개발을 위한 기록

Karl

2022년 1월 19일 • 7 min read

Photo by Jon Tyson / Unsplash

API에 있어 기본이 되는 CRUD API 를 개발하며 FastAPI 에 적응하기 위한 기록을 남긴다.

1. 환경세팅

1) 프로젝트 생성

poetry new apps

poetry 를 통해 프로젝트를 생성하면 아래와 같은 구조를 갖게 된다.

apps
├── pyproject.toml
├── README.rst
├── apps
│   └── __init__.py
└── tests
    ├── __init__.py
    └── test_apps.py

2) 의존성 라이브러리 설치

db는 mysql 을 사용할 것이고, db 비동기 커넥션은 encode.io 에서 개발한 databases 를 사용해보기로 했다. poetry 로 아래와 같이 필요한 라이브러리를 설치 한다.

poetry add 'uvicorn[standard]' fastapi sqlalchemy 'databases[mysql]' python-dotenv mysqlclient

uvicorn[standard]
ASGI Web Server
fastapi
ASGI Framework
databases[mysql]
db 비동기지원 드라이버
sqlalchemy
ORM toolkit
python-dotenv
환경변수 관리

3) DB 도커 컨테이너 세팅

docker run -it --name db -e MYSQL_ROOT_PASSWORD=root -p 3306:3306 -d mariadb

2. DB 접속 환경 개발

fastapi 환경에서 공통적으로 사용할 db 접속환경에 관련된 부분을 개발 해본다. 파일의 위치와 용도는 다음과 같다.

apps
├── .env            # 환경변수 정의
├── pyproject.toml
├── README.rst
├── apps
│   └── __init__.py
│   └── database.py # 디비접속 환경정보 기술
│   └── main.py     # application main
└── tests
    ├── __init__.py
    └── test_apps.py

1) .env

DB_TYPE=mysql
DB_HOST=127.0.0.1
DB_USER=root
DB_PASSWD=root
DB_PORT=3306
DB_NAME=test

.env 파일로 선언한 것을 load_dotenv 함수로 로드해서 os.environ 환경변수로 받아올 것이다.

2) database.py

프로젝트 전체에서 사용할 db 접속환경을 세팅 해주는 부분이다.

# database.py

from databases import Database
from os import environ
from dotenv.main import load_dotenv
from sqlalchemy import create_engine, MetaData

# .env 환경파일 로드
load_dotenv()

# 디비 접속 URL
DB_CONN_URL = '{}://{}:{}@{}:{}/{}'.format(
    environ['DB_TYPE'],
    environ['DB_USER'],
    environ['DB_PASSWD'],
    environ['DB_HOST'],
    environ['DB_PORT'],
    environ['DB_NAME'],
)

# 쿼리를 위한 db 커넥션(비동기) 부분
db_instance = Database(
    DB_CONN_URL, 
    min_size=5,			# 기본 connection 수
    max_size=1000,		# 최대 connection 수
    pool_recycle=500	# mysql 에 설정된 wait_timeout 시간 동안 재요청이 없을 경우 MySQL에서 해당 세션의 연결을 끊어버린다 (https://yongho1037.tistory.com/569)
    )

# 모델 초기화를 위한 db 커넥션 부분
db_engine = create_engine(DB_CONN_URL)
db_metadata = MetaData()

3) main.py

# main.py

from fastapi import FastAPI, Request
from apps.database import db_instance

app = FastAPI(
    title="Memo API",
    description="Memo CRUD API project",
    version="0.0.1"
)

# 서버 시작시 db connect
@app.on_event("startup")
async def startup():
    await db_instance.connect()

# 서버 종료시 db disconnect
@app.on_event("shutdown")
async def shutdown():
    await db_instance.disconnect()

# fastapi middleware, request state 에 db connection 심기
@app.middleware("http")
async def state_insert(request: Request, call_next):
    request.state.db_conn = db_instance
    response = await call_next(request)
    return response

3. Memo, CRUD 기능 개발하기

Memo 기능을 구현하기 위해 아래와 같이 프로젝트 구조를 잡아줬다

apps
├── .env            # 환경변수 정의
├── pyproject.toml
├── README.rst
├── apps
│   └── __init__.py
│   └── database.py # 디비접속 환경정보 기술
│   └── main.py     # application main
│   └── memo
│       └── model.py
│       └── schema.py
│       └── router.py
└── tests
    ├── __init__.py
    └── test_apps.py

1) Model

memo 데이터를 저장할 db 테이블을 정의해보자

# model.py

from datetime import datetime
from apps.database import db_engine, db_metadata
from sqlalchemy import Table, Column, String, Integer, DateTime

memo = Table(
    "memo",
    db_metadata,
    Column("idx", Integer, primary_key=True, autoincrement=True),
    Column("regdate", DateTime(timezone=True), nullable=False, default=datetime.now),
    Column("title", String(255), nullable=False),
    Column("body", String(2048), nullable=False)
)
# 테이블 정보로 테이블 생성한다
memo.metadata.create_all(db_engine)

2) 기본 Router

주요 CRUD 기능을 구현하기 전에 간단한 기능을 router 에 구현하고 main 에 적용하도록 하자

# router.py

from fastapi.routing import APIRouter

memo_router = APIRouter()

@memo_router.get("/test")
async def test():
    return {"say":"hello"}

위에서 간단히 구현한 router를 아래와 같이 main 에 붙이고 기능확인을 한다

# main.py

from fastapi import FastAPI, Request
from apps.database import db_instance
from apps.memo.router import memo_router

app = FastAPI(
    title="Memo API",
    description="Memo CRUD API project",
    version="0.0.1"
)

@app.on_event("startup")
async def startup():
    await db_instance.connect()

@app.on_event("shutdown")
async def shutdown():
    await db_instance.disconnect()

@app.middleware("http")
async def state_insert(request: Request, call_next):
    request.state.db_conn = db_instance
    response = await call_next(request)
    return response

# 이 부분을 추가하여 router 구현로직을 적용
app.include_router(memo_router)

3) Create 기능 구현하기

3-1) Create, Schema

Memo, Create 기능 구현을 위해, 아래와 같이 BaseModel 를 상속받는 MemoCreate 클래스를 생성해줬다.

# schema.py

from datetime import datetime
from pydantic import BaseModel

# Pydantic을 이용한 Type Hinting
class MemoCreate(BaseModel):
    regdate : datetime
    title : str
    body : str

3-2) Create, Router

# router.py

from databases.core import Database
from fastapi.param_functions import Depends
from fastapi.routing import APIRouter
from fastapi.requests import Request

# Memo 스키마 
from apps.memo.schema import MemoCreate
# Memo 모델
from apps.memo.model import memo

memo_router = APIRouter()

# 디비 쿼리를 위한 종속성 주입을 위한 함수
def get_db_conn(request: Request):
    return request.state.db_conn		# middleware 에서 삽입해준 db_conn

# 메모 생성
@memo_router.post("/memo")
async def memo_create(
    req: MemoCreate,					# 요청을 정의한 create 스키마에 맞게 전달 받음
    db: Database = Depends(get_db_conn)
    ):
    
    query = memo.insert()				# insert 쿼리 생성 (feat.sqlalchemy)
    
    # validation 처리가 필요할 듯
    values = req
    
    await db.execute(query, values)		# 쿼리 실행 (feat.databases)
    
    return {**req.dict()}

3-3) 구현기능 확인

4) Select 기능 구현하기
4-1) Select, Schema

# schema.py

from datetime import datetime
from pydantic import BaseModel

# Pydantic을 이용한 Type Hinting
class MemoCreate(BaseModel):
    regdate : datetime
    title : str
    body : str

class MemoSelect(BaseModel):
    idx : int
    regdate : datetime
    title : str
    body : str

4-2) Select, Router

@memo_router.get("/memo/find/{idx}", response_model=MemoSelect)
async def memo_findone(
    idx: int,
    db: Database = Depends(get_db_conn)
):
    query = memo.select().where(memo.columns.idx == idx)
    return await db.fetch_one(query)

@memo_router.get("/memo/list/{page}", response_model=List[MemoSelect])
async def memo_findone(
    page: int = 1,
    limit: int = 10,
    db: Database = Depends(get_db_conn)
):
    offset = (page-1)*limit
    query = memo.select().offset(offset).limit(limit)
    return await db.fetch_all(query)

4-3) 구현기능 확인

5) Delete 기능 구현하기

5-1) Delete, Router

@memo_router.delete("/memo/{idx}")
async def memo_findone(
    idx: int,
    db: Database = Depends(get_db_conn)
):
    query = memo.delete().where(memo.columns.idx == idx)
    await db.execute(query)
    return {"result": "success"}