SQLAlchemy

d4v1d·2022년 3월 11일

개요

FastAPI에서 비동기로 SQL database와 작업할 수 있는 라이브러리 SQLAlchemy의 사용법을 정리한 포스트입니다. 🙊 간단한 예제를 실습하기 위해 로컬 DB를 생성하고 관리할 수 있는 SQLite을 이용했습니다.

Setup

Dependencies

FastAPI와 SQLAlchemy를 사용하기 위해 필요한 dependencies를 설치하고 import합니다.

pip install databases
pip install sqlalchemy
pip install fastapi
pip install aiosqlite

Import & Set up `databases`

필요한 패키지들을 import하고, DATABASE_URL을 지정해서 database 객체를 생성합니다.

main.py

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"	# database의 url

# 만약 PostgreSQL같은 다른 SQL을 사용한다면 다른 DATABASE_URL을 사용해야 함!
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

# DATABASE_URL에 database 객체를 생성
database = databases.Database(DATABASE_URL)

# ...

Create Tables

engine을 생성하고, metadata로부터 테이블들을 만듭니다.

# ...

# SQLAlchemy로부터 메타데이터 생성
metadata = sqlalchemy.MetaData()

# 메타데이터로부터 만들 수 있는 모든 테이블을 생성하여 notes에 저장
notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)

SQLAlchemy 엔진을 생성
engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)

# ...

Create Models

테이블에 생성되어야 하는 (Pydantic)모델 클래스와 SQL 요청에 대한 응답으로 반환될 모델 클래스를 정의합니다.

# ...

# 입력받아서 테이블에 저장될 클래스
class NoteIn(BaseModel):
    text: str
    completed: bool


# 데이터 검색 등의 요청을 받았을 때 response로 반환될 클래스
class Note(BaseModel):
    id: int
    text: str
    completed: bool

# ...

Connect & Disconnect

FastAPI 서버가 열리고 닫힐 때 database와의 연결을 제어합니다.

# ...

app = FastAPI()


# app이 시작할 때
@app.on_event("startup")
async def startup():
    await database.connect()


# app이 종료될 때
@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

# ...

Read Notes

GET method(path operation function)를 추가하여 현재 테이블에 기록된 모든 데이터(List[Note])를 비동기로 읽도록 합니다.

# ...

# /notes에 GET 요청을 보내면 현재 테이블의 모든 데이터를 반환!
@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)
    # return type: List[Note]

# ...

Create Notes

POST method(path operation function)를 추가하여 현재 테이블에 데이터(NoteIn)를 추가합니다.

# ...

@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}
    # **note.dict(): Python unpack 문법
    # note에 "id" key를 추가하여 response로 반환!
    
# ...

Check

main.py

```python
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"

# 만약 PostgreSQL같은 다른 SQL을 사용한다면 다른 DATABASE_URL을 사용해야 함!
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

FastAPI 서버 실행

» uvicorn main:app --reload
INFO:     Will watch for changes in these directories: ['/directory/of/SQLAlchemy/project']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [31874] using watchgod
INFO:     Started server process [31876]
INFO:     Waiting for application startup.
INFO:     Application startup complete.