백엔드 서버는 비즈니스 로직을 담고있을 뿐 아니라, 프론트엔드와 데이터베이스를 연결해주는 브릿지 역할도 합니다. 프론트엔드는 사용자에게 모두 공개 되어 있기 때문에, 바로 데이터베이스에 연결할 수 없습니다. 그래서 백엔드와 REST 통신을 이용하여 데이터 베이스 정보에 접근합니다.
오늘 우리는 데이터베이스를 연결해 볼 것입니다. 매번 항상 하시던 분들은 숨쉬는 것만큼 쉬울 수 있지만, 처음 하시는 분들은 어려우실 수 도 있어서 자세하게 설명 드리겠습니다.
포스팅에 사용된 소스코드 Commit ID : d21f2ea194747194b6a54eaeabf8d11f5045c317
혹시 영상으로 보고 싶으신 분들을 위해 영상도 첨부합니다!
1. 디펜던시 설치
pip install pymysql pip install sqlalchemy
MySQL을 접속하는 드라이버인 PyMySQL과 ORM인 SQLAlchemy를 설치 해 줍니다.
2. conn.py
database 폴더 안에 conn.py 를 아래와 같이 설정했습니다.
from fastapi import FastAPI from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker import logging class SQLAlchemy: def __init__(self, app: FastAPI = None, **kwargs): self._engine = None self._session = None if app is not None: self.init_app(app=app, **kwargs) def init_app(self, app: FastAPI, **kwargs): """ DB 초기화 함수 :param app: FastAPI 인스턴스 :param kwargs: :return: """ database_url = kwargs.get("DB_URL") pool_recycle = kwargs.setdefault("DB_POOL_RECYCLE", 900) echo = kwargs.setdefault("DB_ECHO", True) self._engine = create_engine( database_url, echo=echo, pool_recycle=pool_recycle, pool_pre_ping=True, ) self._session = sessionmaker(autocommit=False, autoflush=False, bind=self._engine) @app.on_event("startup") def startup(): self._engine.connect() logging.info("DB connected.") @app.on_event("shutdown") def shutdown(): self._session.close_all() self._engine.dispose() logging.info("DB disconnected") def get_db(self): """ 요청마다 DB 세션 유지 함수 :return: """ if self._session is None: raise Exception("must be called 'init_app'") db_session = None try: db_session = self._session() yield db_session finally: db_session.close() @property def session(self): return self.get_db @property def engine(self): return self._engine db = SQLAlchemy() Base = declarative_base()
db = SQLAlchemy()
위 db를 앱 시작시 이니셜라이즈 하고 해당 db 만 가지고 사용합니다. 그렇게 싱글턴을 유지합니다.
라인 by 라인 설명은 위 영상에 잘 되어 있습니다.
3. main.py 수정
from dataclasses import asdict from typing import Optional import uvicorn from fastapi import FastAPI from app.database.conn import db from app.common.config import conf from app.routes import index, auth def create_app(): """ 앱 함수 실행 :return: """ c = conf() app = FastAPI() conf_dict = asdict(c) db.init_app(app, **conf_dict) # 데이터 베이스 이니셜라이즈 # 레디스 이니셜라이즈 # 미들웨어 정의 # 라우터 정의 app.include_router(index.router) return app app = create_app() if __name__ == "__main__": uvicorn.run("main:app", host="0.0.0.0", port=8080, reload=True)
위 처럼 변경했습니다. db.init_app 으로 이니셜라이즈 해줍니다.
- [[a.original_name]] ([[a.file_size | fileSizer]])