[SQLAlchemy] shema 사용하기

내가 하고 싶었는데 못했던거 해결해서 까먹을까봐 쓰는 글

실제로 코드를 간략하게 인터넷에서 볼 수 있도록 바꾼거라 실행 안해봄.. (오류가 있을 수 있음)

환경

Python 3.9.13

SQLAlchemy         1.4.44

sample

models.py

from sqlalchemy import TIMESTAMP, Column, String, ForeignKey, text, DATE, TIME, FLOAT, Table
from sqlalchemy.orm import relationship, backref
from sqlalchemy.dialects.mysql import INTEGER, DECIMAL, TINYINT, TEXT
from app.database import Base

class User(Base):
  __tablename__ = "users"
  __table_args__ = {'comment': '사용자'}
  
  id = Column(INTEGER(unsigned=True), primary_key=True, autoincrement=True, nullable=False, comment='사용자 PK')
  username = Column(String(50), nullable=False, comment='사용자 로그인 아이디')
  pw = Column(String(255), nullable=False, comment='비밀번호')
  name = Column(String(50), nullable=False, comment='사용자명')
  phone = Column(String(15), nullable=False, comment='전화번호')
  register = Column(TIMESTAMP(timezone=True), nullable=False, server_default=text('now()'), comment='가입일시')
  withdrawal = Column(TIMESTAMP(timezone=True), nullable=True, comment='탈퇴일시')
  user_type_id = Column(INTEGER(unsigned=True), ForeignKey("code_user_type.id", nullable=True, comment='사용자 타입')

  user_type = relationship("CodeUserType", backref=backref("user"))
  follower = relationship("Follower", backref=backref("user"))
 
class CodeUserType(Base):
  __tablename__ = "code_user_type"
  __table_args__ = {'comment': '사용자 타입'}
  id = Column(INTEGER(unsigned=True), primary_key=True, autoincrement=True, nullable=False, comment='사용자 타입 PK')
  name = Column(String(50), nullable=False, comment='사용자 타입명')
  
class Follower(Base):
  __tablename__ = "follower"
  __table_args__ = (
    db.UniqueConstraint('user_id', 'follower_id', name='unique_follwer'),
  )
  
  user_id = Column(INTEGER(unsigned=True), nullable=False, comment='사용자 id')
  follower_id = Column(INTEGER(unsigned=True), nullable=False, comment='팔로워 사용자 id')
  
  follower = relationship("User", backref=backref("user"))

schemas.py

from pydantic import BaseModel, EmailStr, condecimal
from typing import Optional, List
from datetime import datetime, date, time

Class CodeUserType(BaseModel):
  id: int
  name: str

class UserCreate(BaseModel):
  username: str
  pw: str
  name: str
  phone: str
  user_type_id: int
  
class UserUpdate(BaseModel):
  name: str
  pw: str
  phone: str
  email: EmailStr
  user_type_id: int

class User(BaseModel):
  id: int
  username: str
  name: str
  phone: Optional[str]
  register: datetime  
  user_type_id: int
  
  user_type: CodeUserType

  class Config:
    orm_mode = True
    
class UserRead(User):
  Follow: List[User]

  class Config:
    orm_mode = True

class UserRead2(BaseModel):
  User: UserRead  
  follow_cnt: int

  class Config:
    orm_mode = True

class UserRead3(User):
  Follow: List[User]  
  follow_cnt: int
  
  user_type: CodeUserType

  class Config:
    orm_mode = True

main.py

from fastapi import FastAPI, Request
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
import uvicorn
from routers import users
from loguru import logger
from starlette.routing import Match 
from starlette.concurrency import iterate_in_threadpool
import uuid, contextvars

app = FastAPI()

origins = ["*"]

app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"]
)

app.include_router(user.router)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=3000)

users.py

from fastapi import Depends, status, HTTPException, APIRouter
from sqlalchemy.orm import Session
from database import get_db
import models, schemas, utils, oauth2

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

@router.post("", status_code=status.HTTP_201_CREATED, response_model=schemas.UserRead)
def create_user(new_user: schemas.UserCreate, db: Session = Depends(get_db)):
  """
  `회원가입`
  """
  # password 암호화
  hashed_password = utils.hash(new_user.password)
  new_user.password = hashed_password
  new_user = models.User(**new_user.dict())
  db.add(new_user)
  db.commit()
  db.refresh(new_user)

  return new_user

@router.get("", response_model=schemas.UserRead)
def read_current_user(db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)):
  """
  `내 정보`
  """
  user = db.query(models.User).filter(models.User.id == current_user.id).first()
  
  return user
  
@router.get("", response_model=schemas.UserRead2)
def read_current_user(db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)):
  """
  `내 정보에 무언가 더 하고싶을 경우 User가 보통 객체로 표시된다.
  백엔드에선 편하다 슈슉샥샥 변환하기 편하고 orm이 알아서 다 처리해준다.
  근데 클라이언트쪽에서 사용할때 객체로 들어가서 봐서 불편하다. 그래서 UserRead3 방법도 알려줌..
  `
  """
  user = (db.query(models.User, func.count(models.Follow.id))
          .outerjoin(models.Follow, models.User.id == models.Follow.user_id)
          .filter(models.User.id == current_user.id)
          .group_by(models.User)
          .first())
  
  
  return user

@router.get("", response_model=schemas.UserRead3)
def read_current_user(db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)):
  """
  `내정보 UserRead3
  방법은 2가지가 있다.`
  """
  user_data = (db.query(models.User, models.UserType, func.count(models.Follow.id).label('follow_cnt'))
                 .join(models.UserType, models.User.id == models.UserType.user_id)
                 .outerjoin(models.Follow, models.User.id == models.Follow.user_id)
                 .filter(models.User.id == current_user.id)
          	     .group_by(models.User, models.UserType)
                 .first())
  
  # 1번
  user: schemas.UserRead = dict()
  user.id = user_data.id
  user.username = user_data.username
  user.name = user_data.name
  user.phone = user_data.phone
  user.register = user_data.register
  user.user_type_id = user_data.user_type_id
  user.user_type = user_data.models.UserType
  user.follow_cnt = user_data.models.follow_cnt
  
  # 2번
  # 코드값들은 다 다시 넣어줘야한다는 불편함이 있지만
  # 기본값들은 모두 다 넣을 수 있다.
  farm = schemas.UserRead(
      **user_data.User.__dict__,
      user_type = user_data.UserType,
      follow_cnt = user_data.follow_cnt
  )
  
  
  return user
  
@router.get("/list", response_model=List[schemas.UserRead])
def read_current_user(db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)):
  """
  `회원 목록`
  """
  user_list = db.query(models.User)
  
  return user_list

@router.put("", status_code=status.HTTP_200_OK, response_model=schemas.UserRead)
def update_user(updated_user: schemas.UserUpdate, db: Session = Depends(get_db), current_user: int = Depends(oauth2.get_current_user)):
  """
  `회원 정보 변경`
  """
  user_query = db.query(models.User).filter(models.User.id == current_user.id)
  user = user_query.first()
  
  if not user:
    raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=f"[{id}] 존재하지 않는 사용자입니다.")
  
  # 1번과 2번 중 선택 다른 방법들도 많음

  # 1번
  user_query.update(updated_user.dict(), synchronize_session=False)
  # 2번
  user.name = updated_user.name  
  hashed_password = utils.hash(updated_user.pw)
  new_user.pw = hashed_password  
  user.phone = updated_user.phone
  user.user_type_id = updated_user.user_type_id
  
  db.commit()

  return user