FastAPI と SQL Databases(SQLAlchemy)

この記事はFastAPIでの**SQLAlchemy ORM**の使用方法を解説した以下の公式サイトを読んだ備忘録です。ValidationのためにPydanticクラスとの併用になりますが、全体的に冗長性は否めません。

[FastAPI 公式サイト - SQL (Relational) Databases](https://fastapi.tiangolo.com/ja/tutorial/sql-databases/)

（注意）**SQLModelについて**
また他方、FastAPIの作者自ら**SQLModel**を作成しており今後はそちらの方に移行していくものと予想されます。SQLModelは **Pydantic** と **SQLAlchemy**を統合したものとなっており、コードの冗長性を減らす方向となっています。
[SQLModel 公式サイト](https://sqlmodel.tiangolo.com/)
[**FastAPI SQLModel 入門 - Qiita 2022/08/30　追加**](https://qiita.com/sand/items/f338e67e86d728f793c2)


# 1.ソースコードと解説

ここではORM として **SQLAlchemy ORM**を用い、 SQL (Relational) DatabasesとしてSQLiteを利用します。ちなみにSQLAlchemy　はasync/await と互換性がないことに注意してください。
[SQLAlchemy 1.4 Documentation](https://docs.sqlalchemy.org/en/14/orm/tutorial.html)


```
pip install sqlalchemy
```

バージョンを確認します。

```
>>> import sqlalchemy  
>>> sqlalchemy.__version__                        
'1.4.27'
```


ソースコードツリーです。

```
.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py
```


## 1-1. database.py



```Python:database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()
```

### sessionmaker
**sessionmaker**で**Sessionクラス**のオブジェクトが作られますが、これによってORMがdatabase をハンドルできるようになります。

```
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
```

### Declarative Extensions
**SQLAlchemy ORM** のテーブルとクラスのマッピングは**Declarative Extensions**によって行われます。クラスは**declarative base class**の拡張として定義されることでテーブルとのマッピングを維持できるようになります。

```
Base = declarative_base()
```

## 1-2. models.py

SQLAlchemy ORMによるクラスの定義です。**declarative base class**の拡張として定義されることで自動的にテーブルとのマッピングが維持されます。

```Python:models.py
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")
```

### relationship
以下のようにSQLAlchemy ORM の **relationship** を定義しています。
Userオブジェクト user の user.items は、 Itemのowner_idがuser.idに等しいItemオブジェクトの配列となります。
逆にitem.ownerによって、owner_idによって指し示されるUserオブジェクトにアクセスできるようになります。

```
    items = relationship("Item", back_populates="owner")
    ---
    owner = relationship("User", back_populates="items")
```
 

## 1-3. schemes.py


ここでは**Pydantic**クラスを定義します。（個人的な見解ですが、上で定義したSQLAlchemyクラスを考えれば冗長性は否めません。）

Pydantic はデータ検証のためのPythonライブラリです。Pythonのtype hintが使われます。ですからPydanticクラスはAPIの入り口であるパスオペレーション関数で使われることが一般的です。

APIのCreate/Readで使うクラス定義は小さな差異があることが一般的ですが、初めに両者の共通部分を定義します。

|共通部分|Create|Read|
|---|---|---|
|ItemBase|ItemCreate|Item|
|UserBase|UserCreate|User|


例えばユーザ情報をCreateするときはpasswordが必要ですが、Readするときはpasswordを含む必要がありません。


```Python:schemas.py
from typing import List, Union
from pydantic import BaseModel

class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None

class ItemCreate(ItemBase):
    pass

class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True

class UserBase(BaseModel):
    email: str

class UserCreate(UserBase):
    password: str

class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True
```

## 1-4. crud.py

databaseとのインターフェースを司るユーティリティの集まりです。ここではCRUD(Create, Read, Update, Delete)のCreateとReadのみに絞っています。


```Python:crud.py
from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item
```


ここで**各関数のreturnがSQLAlchemy modelsまたは SQLAlchemy models のリストであり、Pydantic modelsでないことに注意してください。** これは次のmain.pyでもう一度指摘します。


## 1-5. main.py


```Python:main.py
from typing import List

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

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    print(user)
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items
```

まずは前に述べたSQLAlchemy ORM の**declarative base class**の**MetaData.create_all()** methodを使って、テーブルの作成を行います。

```
models.Base.metadata.create_all(bind=engine)
```

### A dependency with yield and try
[Dependencies with yield](https://fastapi.tiangolo.com/ja/tutorial/dependencies/dependencies-with-yield/)
get_db()はdependencyとしてpath operationsから呼ばれ、database sessionを生成し返します(yield)。最終的にAPIのレスポンスを返したときにfinally部分が実行されます。つまりdb.close()が実行されます。
ちなみにyieldを含むget_db()関数は、Pythonのジェネレータ関数です。

```
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()
```



### orm_mode
前に指摘したように、crud.pyの各関数のreturnがSQLAlchemy modelsまたは SQLAlchemy models のリストであり、Pydantic modelsでありません。しかしpath operationではresponse_model=schemas.Userのように、response_modelではPydantic modelsが指定されています。
これはorm_modeの指定により、SQLAlchemy modelsからPydantic modelsの要素が自動に抽出されていることを意味しています。この時通常のようにfiltering や validationが行われます。


今回は以上です。