SQLAlchemy 사용방법

세션 연결하기

sessionmaker : 세션(orm-mapped object를 위한 지속적인 운영을 관리하는 객체)을 만들어 주는 factory. - 엔진의 연결 풀 요청, 세션 객체 연결 및 초기화

declarative_base를 이용하여 table과 db를 매핑 시켜 줌

declarative 클래스는 Table객체를 metadata로 정의 해두어 이를 통해 테이블로 접근 할 수 있게 함.

create_all 로 engine연결된 database에 table을 생성 하거나 연결.

query 수행할 때 session.query와 model.query가 있는데 sqlAlchemy만 사용한다면 session query로만 사용가능. 아래 model.qeury를 선언한 것은 graphQL을 위해서임.

Yield를 이용하여 세션 처리는 실제 transaction을 수행할 때 이뤄지도록 함.

import cx_Oracle
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

dsn = cx_Oracle.makedsn(host="{HOST}", port=${PORT}, sid=${SID})
engine = create_engine("oracle+cx_oracle://${user_id}:${user_pw}@%s" % str(dsn))
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
Base.metadata.create_all(bind=engine)
Base.query = scoped_session(SessionLocal).query_property()

def get_db():
    db = SessionLocal()
    try:
        yield db

    finally:
        db.close()

모델 만들기

앞에 정의된 Base를 통해 database의 테이블과 모델을 매칭 시켜 준다.

필요로 하는 column에 대해 정의 하고, Basemodel을 이용하여 Business Model을 만들어 준다.

유저에게 보여주는 모델이 데이터 베이스의 테이블과 같지 않을 수 있기 때문에 비지니스 모델을 따로 정의 하는 것이 좋다. (개인적인 구현 관점)

ORM모델의 relationship항목은 Item table 을 join하기 위함임.

from typing import Optional

from sqlalchemy.sql.schema import ForeignKey
from pydantic import BaseModel
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.repository.sql_session import Base
from .item import ItemBase

# Orm model for User. it matches with database table "user"
class User(Base):
    __tablename__ = "user"

    name = Column(String(50), primary_key=True)
    department = Column(String(100))
    age = Column(Integer)
    email = Column(String(100), ForeignKey("item.email"))
    phone = Column(String(15), unique=True)
    address = Column(String(200))
    floor = Column(Integer)

    item = relationship("Item", foreign_keys="User.email")

# Base models is for reading and create data
# In case of we have diffrent attributes from database tables, this kind of concept are useful.
class UserBase(BaseModel):
    name: str
    department: str
    age: Optional[int]
    email: Optional[str]
    phone: Optional[str]
    address: Optional[str]
    floor: Optional[int]
    item: Optional[ItemBase]

    # Orm_mode will tell the Pydantic model to read the data even if it is not a dict, but an ORM model
    class Config:
        orm_mode = True
        arbitrary_types_allowed = True

# This class can be used when there are more attributes in creation part
class UserCreate(UserBase):
    pass

ORM을 이용한 CURD 구현

session.query를 이용하여 request요청함.

creation의 경우 commit 이 호출 되어야 실제 transaction이 일어남. 그 외에는 목적에 맞게 orm을 적절하게 사용하면 됨.

import logging
from typing import Text

from app.model.user import User, UserCreate
from sqlalchemy.orm import Session

logger = logging.getLogger("garage")

def get_users(db: Session, skip: int = 0, limit: int = 100):
    logger.info("hello!!")
    return db.query(User).offset(skip).limit(limit).all()

def create_user(db: Session, user: UserCreate):
    logger.info(user)
    db_user = User(
        name=user.name,
        department=user.department,
        age=user.age,
        email=user.email,
        phone=user.phone,
        address=user.address,
        floor=user.floor,
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_name(db: Session, name: str):
    return db.query(User).filter(User.name == name).first()

endpoints 구현하기

FastAPI의 endpoint 와 앞에 작성한 curd를 연결 하면 됨.

from fastapi import APIRouter, Depends, HTTPException
from app.model.user import User, UserBase, UserCreate
from app.repository.sql_session import Base, SessionLocal, engine, get_db
from app.repository import user_curd
from app.model import user
from typing import List
from sqlalchemy.orm import Session
import logging

'''
This Module is for users endpoints.

Client can access this application through GET/POST/PUT/DELETE /users

'''

router = APIRouter()
logger = logging.getLogger("garage")

@router.get("/users", tags=["users"], response_model=List[UserBase],description="Show all user info")
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    logger.info("read users")
    return user_curd.get_users(db, skip=skip, limit=limit)

@router.post("/users", tags=["users"], response_model=UserBase, description="Create a user")
def create_users(user: UserCreate, db: Session = Depends(get_db)):
    logger.info("create a user")
    db_user = user_curd.get_user_by_name(db, name=user.name)
    if db_user:
        logger.error("User post existing user")
        raise HTTPException(status_code=400, detail="Name already registered")
    return user_curd.create_user(db=db, user=user)