sessionmaker로 만든 SessionLocal() 인스턴스가 매 요청마다 yield 되어 커넥션 연결db: get_db = Depends()@staticmethod
dependency-injection 라이브러리
<aside> 💡
DBSession 클래스를 사용해 repository 계층에서 MySQL 서버와 통신
</aside>
디렉토리 구조
user/router.py
from fastapi import APIRouter, Query
from app.user.service import user_service <-----------------------
api = APIRouter(
prefix="/user",
tags=["user"],
responses={
404: {"description": "Not found"},
},
)
@api.get("/test")
def get_all_colleges():
response = user_service.get_all_colleges()
return response
@api.get("/test2")
def get_goal_college_by_rank(rank: int = Query()):
response = user_service.get_goal_college_by_rank(rank)
return response
user/service.py
from typing import List
from app.user.repository import user_repository <------------------------
from app.user.models import GoalCollege
class UserService:
def get_all_colleges(self) -> List[GoalCollege]:
colleges = user_repository.get_goal_colleges()
return colleges
def get_goal_college_by_rank(self, rank: int) -> GoalCollege:
return user_repository.get_goal_college_by_rank(rank)
user_service = UserService()
user/repository.py
from typing import List
from app.core.db.deps import db_session <--------------------------
from app.user.models import GoalCollege
class UserRepository:
def __init__(self) -> None:
self.db = db_session
def get_goal_colleges(self) -> List[GoalCollege]:
with self.db as db:
goal_colleges = db.query(GoalCollege).all()
return goal_colleges
def get_goal_college_by_rank(self, rank: int) -> GoalCollege:
with self.db as db:
goal_college = db.query(GoalCollege).filter(GoalCollege.rank == rank).one()
return goal_college
def deregister(self, user_id: int) -> None:
with self.db() as db:
db.query(User).filter(User.id == user_id).update(
{"deregistered_at": datetime.now(config.KO_TIMEZONE)}
)
db.commit()
user_repository = UserRepository()
core/db/deps.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from app.core.config import config
engine = create_engine(
config.thai_db_url,
# echo=True,
echo_pool="debug",
)
SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)
class DBSession:
def __enter__(self):
self.db = SessionLocal()
return self.db
def __exit__(self, exc_type, exc_value, traceback):
if exc_type:
self.db.rollback()
self.db.close()
db_session = DBSession()