FastAPI, CRUD API 개발을 위한 기록
API에 있어 기본이 되는 CRUD API 를 개발하며 FastAPI 에 적응하기 위한 기록을 남긴다.
일상을 공유하는 소셜 채팅 - SnapTok - Apps on Google Play
It is a social network where you can share your daily life and talk with friends with similar tastes.
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

일상을 공유하는 소셜 채팅 - SnapTok - Apps on Google Play
It is a social network where you can share your daily life and talk with friends with similar tastes.
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
일상을 공유하는 소셜 채팅 - SnapTok - Apps on Google Play
It is a social network where you can share your daily life and talk with friends with similar tastes.
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)

일상을 공유하는 소셜 채팅 - SnapTok - Apps on Google Play
It is a social network where you can share your daily life and talk with friends with similar tastes.
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"}
5-2) 구현기능 확인

일상을 공유하는 소셜 채팅 - SnapTok - Apps on Google Play
It is a social network where you can share your daily life and talk with friends with similar tastes.