FastAPI로 백엔드 CRUD - DB구성 - mysql, sqlalchemy

우분투 폴더 구조

폴더 구조는 server 아래로

config, models, routes, schemas, server(환경 생성시 만들어지는 빈 폴더), tests(환경 생성시 자동 생성)

 

config폴더에서는 sqlalchemy로 Mysql과 연결합니다(db.py)

mysql은 도커에 올려져 있는 상태이며 해당 포트로 열려있게끔 사전에 설정하여야 합니다

from sqlalchemy import create_engine, MetaData
engine = create_engine('mysql+pymysql://root:password@localhost:3306/test')
meta = MetaData()
conn = engine.connect()

 

config 폴더에 db.py에서 연결한 객체를 models 폴더의 user.py에서 import 하여 사용하고 테이블을 구성합니다

from sqlalchemy import Column, Table
from sqlalchemy.sql.sqltypes import INTEGER, String
from config.db import meta, engine
users = Table('users', meta,
              Column('id', INTEGER(), primary_key=True),
              Column('name', String(255)),
              Column('email', String(255)),
              Column('password', String(255)),
              )

meta.create_all(engine)

 

schemas 폴더의 user.py에서 pydantic을 활용해서 data를 validation 하기위한 모델을 만듭니다

from pydantic import BaseModel


class User(BaseModel):
    name: str
    email: str
    password: str

 

다음은 routes 폴더입니다 CRUD의 메인 기능이 담겨 있습니다

from models.user import users
from config.db import conn
from fastapi import APIRouter
from schemas.user import User

user = APIRouter()


@user.get('/')
def fetch_all():
    return conn.execute(users.select()).fetchall()


@user.post('/')
def create_user(user: User):
    conn.execute(users.insert().values(name=user.name,
                 email=user.email, password=user.password))
    return conn.execute(users.select()).fetchall()


@user.put('/{id}')
def update_user(user: User, id: int):
    conn.execute(users.update().values(name=user.name, email=user.email,
                 password=user.password).where(users.c.id == id))
    return conn.execute(users.select().where(users.c.id == id)).first()


@user.delete('/{id}')
def delete_user(id: int):
    conn.execute(users.delete().where(users.c.id == id))
    return conn.execute(users.select().where(users.c.id == id)).first()

 

다음은 server 폴더 내에 index.py 입니다 작은 앱이면 index.py내에서  기능을 처리하겠지만 좀 더 큰 API구조와 다중 폴더 구조를 가지게 될수록 위의 라우팅 기능 활용이 필수 입니다

https://fastapi.tiangolo.com/tutorial/bigger-applications/(공식문서에서의 라우팅 설명)

from fastapi import FastAPI
from routes.user import user
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.include_router(user)

 

테스트 (실행전에 가상환경 진입 해야하고, DB 도커가 running 상태인지 확인해야 합니다)

uvicorn <앱 파일의 이름>:<앱 내의 객체변수이름> --reload

localhost:8000/docs 화면

api swagger

여기에서 각 버튼을 눌러서 api가 정상작동하는지 확인해볼 수 있습니다