최근 팀 프로젝트를 시작했는데 개발자가 나 뿐이라
flutter + fast api 조합으로 프로젝트를 구성하려고 했다.
flutter 는 ios 네이티브 개발을 내가 잘하지 못하는것도 있고 개발 시간을 줄이기 위해 선택했고
fast api 는 복잡한 세팅 없이 빠르게 개발을 진행할 수 있으면서 성능도 준수하다고 하는 fast api 를 선택했다.
파이썬을 잘 알지는 못하지만 개념자체는 다른 백엔드와 다르지 않고 참고할것도 많아서 개발하는데 문제 없을거라고 생각했다.
FastAPI란 무엇인가?
FastAPI는 Python 3.6 이상을 기반으로 한 고성능 웹 프레임워크다. Starlette과 Pydantic 위에 구축되어 있으며, 비동기 프로그래밍을 완벽히 지원하며, 자동 문서화 기능(OpenAPI 기반)이 내장되어 있다. REST API 서버를 빠르고 직관적으로 개발할 수 있도록 돕는 것이 FastAPI의 핵심 철학이다.
FastAPI의 장점
- 빠른 성능: Starlette 기반으로 Node.js보다 빠르거나 유사한 수준의 성능을 보여준다.
- 자동 문서화: Swagger UI와 ReDoc 문서를 자동으로 생성한다.
- 타입 기반의 요청/응답 검증: Pydantic을 활용하여 입력값을 자동으로 검증하고 직렬화한다.
- 비동기 지원: async/await 기반으로 비동기 처리를 자연스럽게 구현할 수 있다.
- 간결한 문법: Flask처럼 직관적인 라우팅 방식과 함께, 더 구조적인 작성이 가능하다.
FastAPI의 단점
- 학습 곡선: Pydantic, async/await, 타입 힌트 등에 익숙하지 않다면 처음 진입 장벽이 있을 수 있다.
- 생태계 규모: Django에 비해 플러그인, ORM 생태계가 작지만 점점 확장되고 있는 중이다.
- WebSocket, 미들웨어 커스터마이징 등 고급 기능은 Starlette 문서를 참고해야 한다.
main
from fastapi import FastAPI
from routers import user
import uvicorn
app = FastAPI()
app.include_router(user.router, prefix="/users", tags=["Users"])
if __name__ == "__main__":
uvicorn.run("main:app", reload=True)
fast api 가 실행되는 main.py 다.
include_router 를 좀 알고 넘어가야하는데 모든 기능을 다 하나의 파일에 때려박아서 만들게 아니라면
/users 라는 url 로 요청할 api 들 이외에도 이것저것 기능별로 나뉠텐데 그것들을 라우터로 분류하고
그 라우터들을 fast api 에서 사용할 수 있게 등록하는 작업이다.
현재는 /users 라는 prefix 로 동작할 user 라우터만 등록된 코드다.
user 라우터
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import List
router = APIRouter()
class User(BaseModel):
id: int
name: str
email: str
fake_db = []
@router.post("/", response_model=User)
def create_user(user: User):
fake_db.append(user)
return user
@router.get("/", response_model=List[User])
def get_users():
return fake_db
@router.get("/{user_id}", response_model=User)
def get_user(user_id: int):
for user in fake_db:
if user.id == user_id:
return user
raise HTTPException(status_code=404, detail="User not found")
@router.put("/{user_id}", response_model=User)
def update_user(user_id: int, updated_user: User):
for index, user in enumerate(fake_db):
if user.id == user_id:
fake_db[index] = updated_user
return updated_user
raise HTTPException(status_code=404, detail="User not found")
@router.delete("/{user_id}")
def delete_user(user_id: int):
for index, user in enumerate(fake_db):
if user.id == user_id:
del fake_db[index]
return {"message": "User deleted"}
raise HTTPException(status_code=404, detail="User not found")
가짜 db 를 예시로 라우터를 구성했다
update_user 를 에시로 들면 "서버주소/users/유저아이디" 이런 식으로 요청하게 될 거다.
물론 put 으로 요청을 해야한다. 그러면 아래의 기능들이 동작하고 response_model 에 적은 User 로 리턴이 된다.
클라이언트 쪽에서 요청하는 걸 간단하게 예를 들어보면
PUT /users/1
updated_user
{
"id": 1,
"name": "최데브",
"email": "choidev@example.com"
}
이렇게 될거다. 예를 알 수 있듯 url 에 포함되지 않고 넘겨주는 데이터는 body 에 json 형태로 서버에 보내줘야한다. 리턴도 json 형태로 자동으로 된다.
ORM
from sqlalchemy import Column, Integer, String
from database import Base
class User(Base): //base 상속
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String)
email = Column(String, unique=True, index=True)
실제 db 테이블과 매핑되는 ORM 이다.
실제 db 연결하기
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()
main.py에서 DB 테이블 생성
from database import Base, engine
from models import user as user_model
# 앱 시작 시 테이블 생성
Base.metadata.create_all(bind=engine)
이 코드를 실행하면 Base 를 상속받고 메모리에 로드된 Base 기반 모든 ORM 모델들을 읽어서 db 자동으로 테이블을 생성해준다.
이미 생성된 상태라면 생성하지 않고 넘어간다.
DB 세션 의존성 주입
# dependencies.py
from database import SessionLocal
from fastapi import Depends
from sqlalchemy.orm import Session
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
실제 DB 기반 CRUD 예시
from fastapi import Depends
from sqlalchemy.orm import Session
from models.user import User as UserModel
from dependencies import get_db
@router.post("/", response_model=User)
def create_user(user: User, db: Session = Depends(get_db)):
db_user = UserModel(**user.dict())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
ORM 을 이용해서 매핑해뒀고 db 의존성도 주입했기 때문에 위 코드를 실행하면 전달받은 use json을 변환해서 db 에 add (insert) 한다.
user.dict()와 UserModel(**user.dict())의 의미
- user.dict()는 Pydantic 모델을 Python의 dict로 변환한다.
- 예: {"id": 1, "name": "최데브", "email": "choidev@example.com"}
- UserModel(**user.dict())는 ** 언패킹을 통해 SQLAlchemy 모델의 생성자에 해당 딕셔너리 값을 할당한다.
간단하게 알아봤다. 정말 얼마 안되는 코드로 간단한 crud 기능은 완성할 수 있다.
비동기 처리나 대규모 데이터 처리를 위해서는 다른것들도 좀 알아봐야하지만 그건 쓰게되는 날에 다시 적어보겠다.
안까먹는다면...
'BackEnd' 카테고리의 다른 글
Fast api 에서 playwright 을 사용할때 NotImplementedError 에러 (0) | 2024.04.28 |
---|