やること
FastAPIベースのアプリにデータベース機能を追加する
前提
FastAPIの入門として、データベースを利用したアプリを作成していきます。
前回の記事はこちら↓
SQLAlchemyとは？
端的に言えば、Pythonでデータベースとやりとりするためのライブラリです。主なメリットは
・SQL文を書かなくてもPythonのコード内でSQLのクエリを生成できる
・ORM(Object Relational Mapping)により、DBのテーブルやレコードをオブジェクトとして扱える
といった点が挙げられます。基本的にはエンジン、モデルのベースクラス、セッションの3つを作成し、セッションオブジェクトにデータをINSERTしたりUPDATEできるといった具合です。これらの構成を簡単に動かしてみると、以下のようになります。
import sqlalchemy
import sqlalchemy.ext.declarative
import sqlalchemy.orm
# エンジンの作成
engine = sqlalchemy.create_engine("sqlite:///:memory:", echo=True)
# モデルのベースクラスを設定
Base = sqlalchemy.ext.declarative.declarative_base()
class Type(Base):
__tablename__ = "types"
id = sqlalchemy.Column(
sqlalchemy.Integer, primary_key=True,autoincrement=True)
name = sqlalchemy.Column(sqlalchemy.String(14))
Base.metadata.create_all(engine)
# セッションの作成
Session = sqlalchemy.orm.sessionmaker(bind=engine)
session = Session()
types = [Type(name="backend"),Type(name="infra"),Type(name="frontend")]
session.add_all(types)
session.commit()
types= session.query(Type).all()
for type in types:
print(type.id, type.name)
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine PRAGMA main.table_info("tests")
INFO sqlalchemy.engine.Engine [raw sql] ()
INFO sqlalchemy.engine.Engine PRAGMA temp.table_info("tests")
INFO sqlalchemy.engine.Engine [raw sql] ()
INFO sqlalchemy.engine.Engine
CREATE TABLE tests (
id INTEGER NOT NULL,
name VARCHAR(14),
PRIMARY KEY (id)
)
INFO sqlalchemy.engine.Engine [no key 0.00177s] ()
INFO sqlalchemy.engine.Engine COMMIT
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine INSERT INTO tests (name) VALUES (?) RETURNING id
INFO sqlalchemy.engine.Engine [generated in 0.00011s (insertmanyvalues) 1/3 (ordered; batch not supported)] ('backend',)
INFO sqlalchemy.engine.Engine INSERT INTO tests (name) VALUES (?) RETURNING id
INFO sqlalchemy.engine.Engine [insertmanyvalues 2/3 (ordered; batch not supported)] ('infra',)
INFO sqlalchemy.engine.Engine INSERT INTO tests (name) VALUES (?) RETURNING id
INFO sqlalchemy.engine.Engine [insertmanyvalues 3/3 (ordered; batch not supported)] ('frontend',)
INFO sqlalchemy.engine.Engine COMMIT
INFO sqlalchemy.engine.Engine BEGIN (implicit)
INFO sqlalchemy.engine.Engine SELECT tests.id AS tests_id, tests.name AS tests_name
FROM tests
INFO sqlalchemy.engine.Engine [generated in 0.00242s] ()
1 backend
2 infra
3 frontend
SQLAlchemyの中でどのようにSQL文が実行されているかが見えて面白いですね。
本題
今回は前回の続きで、SQLAlchemyを使ってデータベースの機能を追加して軽くデータベースを操作してみます。
ディレクトリ構成
先に今回の機能追加後のディレクトリ構成を示しておきます。
fastapi/
│
├── .venv/ # 仮想環境
├── apps/
│ ├── __init__.py
│ ├── app.py # FastAPIアプリケーションのエントリーポイント
│ ├── database.py # データベースの設定（新規作成）
│ ├── crud/ # CRUD関連のロジック
│ │ ├── __init__.py
│ │ ├── models.py # SQLAlchemyのモデル（新規作成）
│ │ └── views.py # APIのルーター
└── .env # 環境変数の設定ファイル
データベースの定義
まず、database.pyにエンジンの作成からベースクラスの設定、セッションの作成まで基本的なデータベース処理を記述します。
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
# エンジンの作成
engine = create_engine("sqlite:///./test.db", connect_args={"check_same_thread": False}, echo=True)
# モデルのベースクラスを設定
Base = declarative_base()
# セッションの作成
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
モデルの定義
次に、crudディレクトリにmodels.pyを作成し、モデルのベースクラスを定義します。
from datetime import datetime
from sqlalchemy import Column, Integer, String, DateTime
from apps.database import Base, get_db
from passlib.context import CryptContext
# Baseを継承してUserモデルを作成
class User(Base):
# テーブル名を設定
__tablename__ = "users"
# カラムを設定
id = Column(Integer, primary_key=True)
username = Column(String, index=True)
email = Column(String, unique=True, index=True)
password_hash = Column(String)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(
DateTime, default=datetime.now, onupdate=datetime.now
)
# パスワードをリセットするためのプロパティ
@property
def password(self):
raise AttributeError("password is not a readable attribute")
# パスワードをセットするためのセッター関数
@password.setter
def password(self, password):
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
self.password_hash = pwd_context.hash(password)
エンドポイントの定義
http://127.0.0.1:8000/sqlで実行できるようapp.pyにエンドポイントを定義します。
from fastapi import Depends
from fastapi import FastAPI
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles
from apps.crud.views import router as crud_router
from apps.crud.models import User
from apps.database import Base, engine, get_db
from pathlib import Path
from sqlalchemy.orm import Session
# FastAPIアプリを作成するためのcreate_app関数
def create_app():
# FastAPIインスタンスを生成
app = FastAPI()
# テーブルの作成
Base.metadata.create_all(bind=engine)
# CRUDルーターを登録
app.include_router(crud_router, prefix="/crud")
# 静的ファイルをマウント
app.mount("/static", StaticFiles(directory="apps/crud/static"), name="static")
return app
app = create_app()
# /sql エンドポイントを定義
@app.get("/sql")
def read_users(db: Session = Depends(get_db)):
# Userテーブルから全てのユーザーを取得
db.query(User).all()
return {"message": "コンソールログを確認してください"}
実行SQL
INFO sqlalchemy.engine.Engine SELECT users.id AS users_id, users.username AS users_username, users.email AS users_email, users.password_hash AS users_password_hash, users.created_at AS users_created_at, users.updated_at AS users_updated_at
FROM users
コメントなど
一旦データベースっぽいものは実装できました。今回やるにあたり、1つのポイントは「循環インポート」が起きないようにディレクトリ構成を考える必要があるということでした。
というのも、当初models.pyにデータベース部分の実装を全部追加すればいいやと思っていたら、app.pyとmodels.pyの間でお互いにモジュールをインポートする必要が生じてしまい、どうもうまくいかなかったんですね。なので、こうしたトラブルを避けるために、実装する以前の段階でアプリの構成をきちんと設計しておく必要があるんだなというのを理解できたのが1つの収穫でした。
Discussion