[Python] SqlAlchemy 유용한 문법 (기초~중급?)

★ sqlalchemy를 더욱 더 활용하는 방법★

 

☆ 본 sample 코드들은 실행은 해보지 않았으니 참고용으로만 사용하시기 바랍니다.☆

※ SQL에 대한 이해도가 낮은 경우 이해가 안될 수 있습니다. ※

> 질문은 언제나 환영합니다 <

 

이해를 돕기 위해, 아래 내용은 도서관 api(?)를 기준으로 작성해 보았습니다.

 

1. 객체로 내보내기 (response_model)

used

모델에 맞춰서 만들 필요없이 내가 필요한 것들만 정의해놓으면 된다.

재사용할 수 있기 때문에 좋으며, orm_mode를 사용하면 하위 객체를 불러와줘서 좋다.

class Book(BaseModel):
  id: int
  name: str

@app.get('/books', response_model=List[Book])
def read_books():
    books = session.query(Book).all()  # Book 모델의 모든 책 정보를 조회
    return books

sample

이해를 돕기 위해, 책 리스트와 책을 빌려갔던 이력을 모두 담고 있는 데이터베이스 모델을 가정

 

models.py

데이터 베이스에 대한 정의

from sqlalchemy import Column, Integer, String, Date, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship

Base = declarative_base()

class Book(Base):
    __tablename__ = 'book'
    __table_args__ = {'comment': '책 테이블'}

    id = Column(Integer, primary_key=True, index=True, comment='책 ID')
    title = Column(String(50), comment='책 제목')
    author = Column(String(50), comment='저자')
    published_date = Column(Date, comment='출판일')
    isbn = Column(String(20), comment='ISBN')

class BorrowedBook(Base):
    __tablename__ = 'borrowed_book'
    __table_args__ = {'comment': '대출한 책 테이블'}

    id = Column(Integer, primary_key=True, index=True, comment='대출 ID')
    book_id = Column(Integer, ForeignKey('book.id'), comment='책 ID (FK)')
    borrower_name = Column(String(50), comment='대출자 이름')
    borrow_date = Column(Date, comment='대출일')
    return_date = Column(Date, comment='반납일')

    book = relationship('Book', backref='borrowed_books')

schemas.py

response 또는 request 데이터에 대한 정의

from typing import List, Optional
from pydantic import BaseModel

class BookBase(BaseModel):
    title: str
    author: str
    published_date: str
    isbn: str

class Book(BookBase):
    id: int  # 책 ID

    class Config:
        orm_mode = True

class BorrowedBookBase(BaseModel):
    borrower_name: str  # 대출자 이름
    borrow_date: str  # 대출일
    return_date: Optional[str]  # 반납일 (선택사항)

class BorrowedBook(BorrowedBookBase):
    id: int  # 대출 ID
    book: Book  # 대출한 책 정보

    class Config:
        orm_mode = True

api

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
import models, schemas

app = FastAPI()

@app.get('/books', response_model=List[schemas.Book])
def read_books():
    books = session.query(Book).all()  # Book 모델의 모든 책 정보를 조회
    return books

@app.get('/borrowed_books', response_model=List[schemas.BorrowedBook])
def read_borrowed_books():
    borrowed_books = session.query(BorrowedBook).join(Book).all()  # BorrowedBook 모델과 Book 모델을 조인하여 모든 대여 기록 조회
    return borrowed_books

@app.get('/books/{book_id}/borrowers', response_model=List[schemas.BorrowedBook])
def read_borrowers(book_id: int, db: Session = Depends(get_db)):
    book = db.query(Book).filter(Book.id == book_id).first()  # book_id를 기준으로 Book 모델에서 책 정보 조회
    if not book:
        raise HTTPException(status_code=404, detail="Book not found")
    borrowers = db.query(BorrowedBook).filter(BorrowedBook.book_id == book_id).all()  # book_id를 기준으로 BorrowedBook 모델에서 대여 기록 조회
    return borrowers

 

2. body 객체로 받기

used

class BookSearchOption(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    published_date: Optional[str] = None
    isbn: Optional[str] = None

@app.post('/books/search', response_model=List[schemas.Book])
def search_books(option: BookSearchOption, db: Session = Depends(get_db)):
    query = db.query(Book)
    if option.title:
        query = query.filter(Book.title.like(f"%{option.title}%"))
    if option.author:
        query = query.filter(Book.author.like(f"%{option.author}%"))
    if option.published_date:
        query = query.filter(Book.published_date == option.published_date)
    if option.isbn:
        query = query.filter(Book.isbn == option.isbn)
    books = query.all()
    return books

sample

from fastapi import FastAPI, HTTPException, Depends
from sqlalchemy.orm import Session
from pydantic import BaseModel
import schemas

app = FastAPI()

class BookSearchOption(BaseModel):
    title: Optional[str] = None
    author: Optional[str] = None
    published_date: Optional[str] = None
    isbn: Optional[str] = None

@app.post('/books/search', response_model=List[schemas.Book])
def search_books(option: BookSearchOption, db: Session = Depends(get_db)):
    """
    도서를 검색하는 엔드포인트입니다.
    :param option: 검색 옵션 (title, author, published_date, isbn)
    :param db: 데이터베이스 세션
    :return: 도서 검색 결과
    """
    query = db.query(Book)
    if option.title:
        query = query.filter(Book.title.like(f"%{option.title}%"))
    if option.author:
        query = query.filter(Book.author.like(f"%{option.author}%"))
    if option.published_date:
        query = query.filter(Book.published_date == option.published_date)
    if option.isbn:
        query = query.filter(Book.isbn == option.isbn)
    books = query.all()
    return books

 

3. subquery

used

1. ,subquery()를 붙여 subquery인것을 정의

2. subquerydml 안에 있는 객체를 불러올때는 subquery.c.column 으로 불러올 수 있다.

sub_query = db.query(BorrowedBook.book_id, func.count(BorrowedBook.id).label('count')).group_by(BorrowedBook.book_id).subquery()
borrowed_books = db.query(BorrowedBook).join(Book).add_columns(sub_query.c.count).all()

sample1

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from sqlalchemy import select, exists
import schemas

app = FastAPI()

@app.get('/books/available', response_model=List[schemas.Book])
def get_available_books(db: Session = Depends(get_db)):
    """
    대출 가능한 도서를 조회하는 엔드포인트입니다.
    :param db: 데이터베이스 세션
    :return: 대출 가능한 도서 목록
    """
    # 대출 중인 도서 ID 서브쿼리
    subquery = select(BorrowedBook.book_id).where(BorrowedBook.return_date.is_(None)).subquery()

    # 대출 중이지 않은 도서 목록 조회
    books = db.query(Book).filter(~exists(subquery.select())).all()
    return books

sample2

# 서브쿼리에서 반환한 필드를 schemas.BorrowedBook 모델에 추가하여 함께 반환하는 예시
from sqlalchemy import func

sub_query = db.query(BorrowedBook.book_id, func.count(BorrowedBook.id).label('count')).group_by(BorrowedBook.book_id).subquery()

# schemas.BorrowedBook 모델에 sub_query.c.count 필드를 추가
class BorrowedBook(BaseModel):
    id: int
    book_id: int
    borrower_id: int
    borrowed_date: str
    count: int  # sub_query.c.count 필드 추가

# API 핸들러에서 add_columns 메서드를 사용하여 서브쿼리 결과를 함께 반환
borrowed_books = db.query(BorrowedBook).join(Book).add_columns(sub_query.c.count).all()
return borrowed_books

 

4.