Repository navigation
postgresql 연동 #4
Description
Activity
sqlalchemy orm사용
migration 필요. 우선 수동으로
from database import Base, engine from domain.project.models import Project if __name__=="__main__": print("Creating database ...") try: Base.metadata.create_all(engine) except Exception as e: print(f"[-] Error create database: {e}")
- db의 세션을 얻기 위해 sessionmaker을 이용하자
- autucommit은 비활성화
from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL = f"postgresql://{cnf.POSTGRESQL_USER}:{cnf.POSTGRESQL_PASSWORD}@{cnf.POSTGRESQL_HOST}/{cnf.POSTGRESQL_DB}" engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=True) Base = declarative_base() SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False) db = SessionLocal()
db 세션이 필요한 곳은 DI를 이용
def get_db(): db = SessionLocal() try: yield db # 세션 쓰레드 풀에서 세션을 얻어오고 finally: db.close() # 세션을 반납 from sqlalchemy import Session from fastapi import Depends def get_demo(db: Session = Depends(get_db)): db.query(...)
- 입문자용 강의는
https://youtu.be/d_ugoWsvGLI가 좋았다. - 익숙해지면 객체지향관점에서 다루는 강의는
https://youtu.be/gQTRsZpR7Gw가 좋았다.
- 입문자용 강의는
- 폴더 구조
- models.py
- schemas.py
- crud.py
- 폴더 구조
시간(생성, 수정)은 mixin으로 하는 방법이 좋은 것 같다.
from datetime import datetime from sqlalchemy import Column, DateTime from sqlalchemy.orm import declarative_mixin from pytz import timezone KST = timezone("Asia/Seoul") @declarative_mixin class TimeStamp: created_at = Column(DateTime, default=datetime.now(tz=KST), nullable=False) updated_at = Column(DateTime, default=datetime.now(tz=KST), nullable=False)
from database import Base from sqlalchemy import String, Integer, Column from domain.db_mixins import TimeStamp class Project(TimeStamp, Base): __tablename__ = "project" id = Column(Integer, primary_key=True) name = Column(String(255), nullable=False, unique=True) description = Column(String(255)) def __repr__(self) -> str: return f"<Project name={self.name} description={self.description}"
참고자료