FastAPIとSQLAlchemyの基本的な説明
FastAPIは、Python 3.6以降のバージョンで動作する、現代的で高速（高性能）なWebフレームワークです。FastAPIは、Pythonの型ヒントを使用してAPIのパラメータを宣言し、自動的にリクエストのバリデーション、シリアライゼーション、ドキュメンテーションを行います。
一方、SQLAlchemyは、Pythonでリレーショナルデータベースを操作するためのライブラリです。SQLAlchemyは、SQLのコマンドをPythonのコードで表現することができ、データベースとのやり取りを抽象化します。SQLAlchemyは、データベースの接続管理、トランザクション管理、データモデリングなど、データベース操作に必要な多くの機能を提供します。
- FastAPIとSQLAlchemyを組み合わせることで、Pythonで効率的かつ安全にWeb APIとデータベースを操作することが可能になります。
環境構築と必要なパッケージのインストール
FastAPIとSQLAlchemyを使用するための環境を構築し、必要なパッケージをインストールする手順は以下の通りです。
まず、新しいPython環境を作成します。ここでは、venvを使用しています。
python3 -m venv env
source env/bin/activate
次に、FastAPIとSQLAlchemyをインストールします。
pip install fastapi
pip install sqlalchemy
これで、FastAPIとSQLAlchemyを使用するための環境が整いました。
次に、FastAPIを動作させるためのASGIサーバーであるuvicornをインストールします。
pip install uvicorn
これで、FastAPIアプリケーションをローカルで実行するための準備が整いました。
最後に、データベースとの接続を管理するためのライブラリであるsqlalchemy_utilsをインストールします。
pip install sqlalchemy_utils
- 以上で、FastAPIとSQLAlchemyを使用するための環境構築と必要なパッケージのインストールが完了しました。次のステップでは、これらのツールを使用してデータベース接続を設定します。それについては次のセクションで説明します。
データベース接続の設定
FastAPIとSQLAlchemyを使用してデータベースに接続するための設定は以下の通りです。
まず、SQLAlchemyのcreate_engine関数を使用してデータベースエンジンを作成します。この関数はデータベースのURLを引数に取ります。
from sqlalchemy import create_engine
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL)
上記の例では、SQLiteデータベースを使用していますが、他のデータベース（PostgreSQL、MySQLなど）を使用する場合は、適切なデータベースURLを指定します。
次に、SQLAlchemyのSessionLocalクラスを使用してデータベースセッションを作成します。このセッションを使用してデータベースとのやり取りを行います。
from sqlalchemy.orm import sessionmaker
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
- これで、FastAPIとSQLAlchemyを使用してデータベースに接続するための設定が完了しました。次のステップでは、これらの設定を使用してデータモデルを作成します。それについては次のセクションで説明します。
モデルの作成
FastAPIとSQLAlchemyを使用してデータモデルを作成する手順は以下の通りです。
まず、SQLAlchemyのBaseとdeclarative_baseを使用してベースモデルを作成します。
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
次に、このベースモデルを継承して具体的なデータモデルを作成します。以下に、Userという名前のモデルを作成する例を示します。
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String, index=True)
email = Column(String, unique=True, index=True)
hashed_password = Column(String)
このUserモデルは、usersという名前のテーブルを表し、id、name、email、hashed_passwordという4つのカラムを持っています。
- これで、FastAPIとSQLAlchemyを使用してデータモデルを作成するための設定が完了しました。次のステップでは、これらの設定を使用してCRUD操作を実装します。それについては次のセクションで説明します。
CRUD操作の実装
FastAPIとSQLAlchemyを使用してCRUD（Create, Read, Update, Delete）操作を実装する手順は以下の通りです。
まず、データベースに新しいレコードを作成（Create）する関数を作成します。
from sqlalchemy.orm import Session
from . import models, schemas
def create_user(db: Session, user: schemas.UserCreate):
hashed_password = get_password_hash(user.password)
db_user = models.User(email=user.email, hashed_password=hashed_password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
次に、データベースからレコードを読み取る（Read）関数を作成します。
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
次に、データベースのレコードを更新（Update）する関数を作成します。
def update_user(db: Session, user: schemas.UserUpdate, user_id: int):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
db_user.email = user.email
db_user.hashed_password = get_password_hash(user.password)
db.commit()
db.refresh(db_user)
return db_user
最後に、データベースからレコードを削除（Delete）する関数を作成します。
def delete_user(db: Session, user_id: int):
db_user = db.query(models.User).filter(models.User.id == user_id).first()
db.delete(db_user)
db.commit()
return db_user
- これで、FastAPIとSQLAlchemyを使用してCRUD操作を実装するための設定が完了しました。次のステップでは、これらの設定を使用してエンドポイントを作成します。それについては次のセクションで説明します。
エンドポイントの作成
FastAPIとSQLAlchemyを使用してエンドポイントを作成する手順は以下の通りです。
まず、FastAPIのFastAPIクラスを使用してアプリケーションを作成します。
from fastapi import FastAPI
app = FastAPI()
次に、作成したCRUD操作を使用してエンドポイントを作成します。以下に、ユーザーを作成するエンドポイントとユーザーを取得するエンドポイントを作成する例を示します。
from . import crud, models, schemas
from sqlalchemy.orm import Session
from fastapi import Depends, FastAPI, HTTPException
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
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)):
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/{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
- これで、FastAPIとSQLAlchemyを使用してエンドポイントを作成するための設定が完了しました。次のステップでは、これらの設定を使用してテストを実行します。それについては次のセクションで説明します。
テストの実行
FastAPIとSQLAlchemyを使用して作成したエンドポイントのテストを実行する手順は以下の通りです。
まず、FastAPIのTestClientを使用してテストクライアントを作成します。
from fastapi.testclient import TestClient
client = TestClient(app)
次に、作成したテストクライントを使用してエンドポイントをテストします。以下に、ユーザーを作成するエンドポイントとユーザーを取得するエンドポイントをテストする例を示します。
def test_create_user():
response = client.post(
"/users/",
json={"email": "[email protected]", "password": "testpassword"},
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert "id" in data
user_id = data["id"]
def test_read_user(user_id: int):
response = client.get(f"/users/{user_id}")
assert response.status_code == 200
data = response.json()
assert data["email"] == "[email protected]"
assert data["id"] == user_id
- これで、FastAPIとSQLAlchemyを使用してテストを実行するための設定が完了しました。これらのテストを使用して、エンドポイントが正しく機能していることを確認できます。
0件のコメント