장고(django)보다 쉽게 FastAPI로 간단한 CRUD 만들기

fastapi review

기본 설계

• fastapi로 CRUD 구현
• postgres를 사용할것임
• orm은 SQLALCHEMY 사용할것임

순서

• pip install fastapi[all] 패키지 설치
• pip install sqlalchemy 패키지 설치
• main.py 함수 생성
• hello fastapi 구현
• fastapi 서버와 postgres 서버 연결
• pgAdmin으로 수동으로 DB 생성

postgres와 SQLALCHMY

create_engine, sessionmaker 생성

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base


URL = "postgresql://아이디:패스워드@서버주소/DB이름"

engine = create_engine(URL,echo=False)
sessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

모델 Table 생성

import sqlalchemy
from sqlalchemy import Column, String, Integer, Boolean, DateTime
from .database import Base


class User(Base):
    __tablename__ = "user_dbs"

    id = Column(Integer,primary_key=True,index=True)
    username = Column(String, index=True)
    age = Column(Integer,index=True)
    createAt = Column(DateTime(timezone=True),server_default=sqlalchemy.sql.func.now())

main.py DB 부분 추가

from fastapi import FastAPI
from database import Base, engine, sessionLocal

Base.metadata.create_all(bind=engine)
app = FastAPI()

def get_db():
    db = sessionLocal()
    try:
        yield db
    finally:
        db.close()

Read 부분

class ResponseUser(BaseModel):
    id:int
    username : str
    age :int
    createAt:datetime.datetime

    class Config:
        orm_mode =True

@app.get("/user",response_model=List[ResponseUser])
async def responseUser(db:Session=Depends(get_db)):
    users = db.query(User).all()
    return users

Create 부분

# Create 
class InputUser(BaseModel):
    username : str
    age :int

    class Config:
        orm_mode =True


@app.post("/input",response_model=ResponseUser)
async def inputUser(inputuser:InputUser, db:Session=Depends(get_db)):
    user = User(username=inputuser.username,age=inputuser.age)
    db.add(user)
    db.commit()
    return user

Delete

# Delete
@app.delete("/delete/{id}")
async def deleteUser(id,db:Session=Depends(get_db)):
    user = db.query(User).filter(User.id==id).first()
    db.delete(user)
    db.commit()
    return "삭제완료"

Update

class UpdatetUser(BaseModel):
    id: int
    username : str
    age :int

    class Config:
        orm_mode =True


@app.post("/update/",response_model=ResponseUser)
async def updateUser(inputuser:UpdatetUser, db:Session=Depends(get_db)):

    user = db.query(User).filter(User.id==inputuser.id).first()

    user.username = inputuser.username
    user.age = inputuser.age

    db.add(user)
    db.commit()
    return user