old/WebProject

[FastAPI] 3. FastAPI - MariaDB(MySQL) Restful API / CRUD

뒷골목프로그래머 2022. 9. 26. 17:37
반응형

 지난 포스트에서 SQLalchemy를 활용하여 FastAPI와 MariaDB(MySQL)을 연동하는 방법과 Docker Compose를 활용하여 실행시키는 방법까지 알아보았습니다. 본 포스트에서는 Restful API를 통해 간단하게 CRUD를 어떻게 구현하였는지 살펴보겠습니다.

 

본 포스팅의 내용을 학습하기에 앞서 아래 내용이 이미 구현, 숙지되어 있음을 가정합니다.

 

 

1. Project 개요

  • 프로젝트 구조 
    project_root
│
├── docker-compose.yml
├── backend
│   └── app
│       ├── main.py
│       ├── Dockerfile
│       ├── requirements.txt
│       │
│       ├── business    # 비즈니스 로직 구현 (여러 Service 통합 지점)
│       ├── config      # db연결
│       ├── controllers # API Router
│       ├── models      # Table Schema
│       ├── schemas     # Request, Response Schema
│       └── services    # CRUD 구현
│
└── db # DB 설정
│   ├── conf.d
│   ├── data
│   └── initdb.d
│
├── frontend # 추후 반영
├── nginx # 추후반영
  • 예제 테이블 구조
    • 테이블 명 : Users
    • SchemasUsers Table Properties
  • API 정의
    HTTP Method CRUD Method End point
    POST Create /users
    GET Read (List) /users
    GET Read (Single) /users/{user_id}
    PUT Update /users/{user_id}
    DELETE Delete /users/{user_id}

2. API Route

  controllers 디렉토리 아래에 API Collection 별로 파일을 생성하여 라우팅 합니다. 예를 들어 '/users' 라는 collection으로 API를 작성하고자 한다면, controllers/users.py 를 작성하고 main.py에 include 시켜줍니다. (아래 예제에서 import 된 schemas, business 부분은 뒤 편에서 작성하겠습니다.)

# app/controllers/users.py

from typing import List
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session

from config.database import get_db
from schemas.users import UsersCreateItem, UsersUpdateItem, UsersReadItem
from businesses.users import UsersBusiness


router = APIRouter(
  prefix='/users',
  tags=['Users']
)


@router.post("/", response_model=UsersReadItem)
def create(user: UsersSchema, db: Session = Depends(get_db)):
    return UsersBusiness().create(db=db, user=user)


@router.get("/", response_model=List[UsersReadItem])
def get_list(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return UsersBusiness().get_list(db=db, skip=skip, limit=limit)


@router.get("/{user_id}", response_model=UsersReadItem)
def get(user_id: str, db: Session = Depends(get_db)):
    return UsersBusiness().get(db=db, user_id=user_id)


@router.put("/{user_id}")
def update(user_id: str, req: UsersUpdateItem, db: Session = Depends(get_db)):
    return UsersBusiness().update(db=db, user_id=user_id, user=req)


@router.delete("/{user_id}")
def delete(user_id: str, db: Session = Depends(get_db)):
    return UsersBusiness().delete(db=db, user_id=user_id)
# /app/main.py

from fastapi import FastAPI

from app.config.database import engine, Base
from app.controllers import users

Base.metadata.create_all(bind=engine)

app = FastAPI()
app.include_router(users.router)

 

3. Model

  sqlalchemy를 활용해 예제를 위해 정의한 Users 테이블 정보와 같이 model을 정의합니다.

# app/models/users.py


from sqlalchemy import Column, String
from database import Base


class UserInfo(Base):
    __tablename__ = 'Users'

		user_id = Column(String, primary_key=True, index=True)
    password = Column(String)
    email = Column(String)
    nickname = Column(String, nullable=True)

 

4. Schema

  FastAPI는 pydantic을 활용해 Data를 검증 하는 것을 권장합니다. 따라서 API 호출 시 Request, Response Schema를 미리 설정하고 데이터 검증에 활용합니다.

# app/schemas/users.py


from pydantic import BaseModel, Field, EmailStr


class UsersCreateItem(BaseModel):
    user_id: str = Field(
        default=None, title='사용자 아이디'
    )
    email: EmailStr = Field(
        default=None, title='이메일 형식의 아이디'
    )
    password: str = Field(
        min_length=8, max_length=20, description='비밀번호는 8~20자의 영문, 숫자, 특수문자 조합으로 작성해주세요.'
    )
    nickname: str = Field(
        min_length=2, max_length=8, description='닉네임은 2~8글자의 영문 또는 한글로 만들어주세요.'
    )

    class Config:
        orm_mode = True

class UsersUpdateItem(BaseModel):
    email: EmailStr | None
    nickname: str | None

    class Config:
        orm_mode = True

class UsersReadItem(BaseModel):
    email: EmailStr
    user_id: str
    nickname: str

    class Config:
        orm_mode = True

 

5. DB Data 처리

 services에서 sqlalchemy orm을 활용해 실제 CRUD를 처리합니다.

# app/services/users.py

import datetime
from sqlalchemy.orm import Session

from models.users import Users
from schemas.uesers import UsersCreateItem, UsersUpdateItem


class UserService:
		def create(self, db: Session, user: UsersCreateItem):
		    db_user = Users(
		        user_id=user.user_id,
		        email=user.email,
		        password=user.password,
		        nickname=user.nickname
		    )
		    db.add(db_user)
		    db.commit()
		    db.refresh(db_user)
		    return db_user
		
		def read_list(self, db: Session, skip: int = 0, limit: int = 100):
		    return db.query(Users).offset(skip).limit(limit).all()
		
		def read(self, db: Session, user_id: str):
		    result = db.query(Users).filter(Users.user_id == user_id).first()
		    return result
		
		def update(self, db: Session, user_id: str, user: UsersUpdateItem):
		    db_user = db.query(Users).filter(Users.user_id == user_id).update({
		            Users.nickname: user.nickname,
		            Users.email: user.email,
		        }
		    )
		    db.commit()
		    return db_user
		
		def delete(self, db: Session, user_id: str):
		    db_user = db.query(Users).filter(Users.user_id == user_id).first()
		    db.delete(db_user)
		    db.commit()
		    return db_user

 

6. Business logic 구현

 저는 Controllers(API Router) 와 Services 사이에 Businesses 라는 layer를 하나 추가 하였습니다. 이유는 다음과 같습니다.

  • Restful API는 간단한 CRUD만 있는 것이 아니다.
  • 다양한 db에서 불러온 data는 그대로 Client로 전달되는 것이 아니라 비즈니스 로직을 거쳐서 가공된다.
  • 여러 Service를 필요로 하는 API 호출 시 Businesses 계층에서 데이터를 가공할 수 있음
  • validation check 등의 작업을 Businesses Layer에서 수행
  • Controllers(API Router)와 Service Layer는 각각 라우팅, DB Data 처리 본연의 임무만 수행
# app/businesses/users.py

from sqlalchemy.orm import Session
from fastapi import HTTPException

from services.users import UsersService
from schemas.users import UsersCreateItem, UsersUpdateItem


class UsersBusiness:
    def create(self, user: UsersCreateItem, db: Session):
        return UsersService().create(db=db, user=user)

    def get_list(self, db: Session, skip: int = 0, limit: int = 100):
        return UsersService().get_list(db=db, skip=skip, limit=limit)

    def get(self, user_id: str, db: Session):
        db_user = UsersService().get(db, user_id)
        if db_user is None:
          raise HTTPException(status_code=404, detail='User not found')
        return db_user

    def update(self, user_id: str, user: UsersUpdateItem, db: Session):
        UsersService().update(db, user_id, user)
        return {"messages": "success"}

    def delete(self, user_id: str, db: Session):
        UsersService().delete(db, user_id)
        return {"messages": "success"}

 

7. 맺음말

 CRUD 예제를 통해 sqlalchemy의 간단한 사용법과 FastAPI 프로젝트 구조를 살펴보았습니다. 프로젝트 구조의 경우, 제가 개발 초기에 가장 많이 고민하고 변경한 부분 입니다. 여러 개의 계층이 생기게 되면서 동일한 인자를 계층마다 넘겨줘야 하는 등의 번거로움도 있지만 계층별 역할을 명확히 함으로써 유지보수성과 가독성을 높일 수 있다고 생각합니다. 본 포스팅을 통해 Backend 프로젝트 구조를 잡는 데 어려움이 있으신 분들에게 약간이나마 도움이 되시길 바랍니다.

 본 포스팅에서 User 정보의 CRUD를 예제로 다뤘습니다. CRUD만을 설명하려다보니 회원가입이라는 특성에 맞지 않게 되었습니다. 첫 째로 create_user에서 인자로 받는 값은 FormData 형태로 넘어와야 합니다. 둘째로 DB에 password가 plain text로 입력되는 일은 절대로 있어서는 안될 것 입니다. 그리하여, 다음 포스팅에는 앞서 제기한 두 가지 문제 해결을 포함한 FastAPI 회원가입 예제를 작성해보겠습니다. 부족한 글 끝까지 읽어주셔서 감사합니다.

반응형
저작자표시 비영리

'old > WebProject' 카테고리의 다른 글

[FastAPI] 2. FastAPI - MariaDB(MySQL) Docker Compose 로 연동, Project 구조 세팅  (2) 2022.09.26
[FastAPI] 0. 프로젝트 개요  (0) 2022.09.15

'old/WebProject'의 다른글

  • 현재글[FastAPI] 3. FastAPI - MariaDB(MySQL) Restful API / CRUD

관련글

댓글

티스토리툴바