FastAPIを用いたデータベースCRUD操作の実装

by mr_osson 投稿日:

FastAPIとは

FastAPIは、Pythonの非常に高速（高性能）、使いやすい、モダンな、高速（クイック）なWebフレームワークです。これは、Python 3.6以降の型ヒントに基づいています。

FastAPIは、APIを構築するための最新のベストプラクティスと最適なツールを組み合わせています。これには、Pythonの型ヒント、ASGI、Starlette、およびPydanticが含まれます。

FastAPIの主な特徴は次のとおりです：

  • 高速: NodeJSやGoと同等の非常に高速なパフォーマンス（StarletteとPydanticのおかげで）。
  • 高速なコーディング: 開発者の生産性を約2倍から3倍に向上させます。開発時間を大幅に短縮し、バグを減らし、開発者が直感的に何をすべきかを理解するのを助けます。
  • 少ないバグ: デベロッパーが新しいバグを導入する可能性を減らします。システムが明確であるため、バグが発生しにくくなります。
  • 直感的: 素晴らしいエディタのサポート。自動補完がどこでも機能します。これにより、必要な時間が大幅に短縮され、バグが少なくなります。
  • 簡単: APIの設計が簡単で直感的になります。すべてのパラメータの型が明示的であるため、非常に直感的でユーザーフレンドリーなAPIが得られます。
  • 短縮形式、長縮形式: パラメータを宣言するための短縮形式と長縮形式の両方を使用できます。パラメータはパス、クエリ、クッキー、ヘッダーなどに宣言できます。
  • 堅牢: コードが正しく動作していることを確認するためのテストとカバレッジを取得するためのツールが用意されています。
  • SQLAlchemy、MongoDB、Redis、などとの互換性: SQLAlchemy、MongoDB、Redis、などとの統合をサポートしています。
  • OAuth2とJWT、HTTP Basic Authなどのセキュリティと認証: OAuthやJWT、HTTP Basic Authなどのセキュリティスキームを統合し、自動的に対話型のOAuth2ログインシステムを提供します。
  • 依存性注入: 依存性注入システムを組み込んでいます。これにより、依存関係が自動的に解決され、依存関係が必要な場所で使用できます。
  • 自動対話型APIドキュメンテーション: 自動的に対話型APIドキュメンテーションと探索Webユーザーインターフェースが生成されます。

これらの特徴により、FastAPIは現代のWebアプリケーションとマイクロサービスの開発に非常に適しています。また、データサイエンスと機械学習の分野での使用にも適しています。これは、FastAPIがPython 3.6以降の型ヒントを使用しているため、データの検証と直列化が容易であり、エラーを早期に検出できるからです。また、FastAPIはASGIに準拠しているため、WebSocketsやGraphQLなどの他のプロトコルもサポートしています。これにより、リアルタイムの通信や他の高度なWebアプリケーション機能を実装することが可能になります。

FastAPIでのデータベース接続

FastAPIを使用してデータベースに接続する方法はいくつかありますが、ここではSQLAlchemyと一緒に使用する方法を説明します。SQLAlchemyはPythonで最も人気のあるSQLツールキットであり、ORM（Object Relational Mapper）を提供しています。

まず、SQLAlchemyと必要なライブラリをインストールします。

pip install sqlalchemy async-exit-stack async-generator

次に、データベースとの接続を設定します。この例では、SQLiteデータベースを使用しますが、PostgreSQLやMySQLなど他のデータベースを使用することも可能です。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"

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

Base = declarative_base()

上記のコードでは、データベースURLを指定し、create_engine関数を使用してデータベースエンジンを作成しています。sessionmaker関数は、データベースセッションを作成するための工場（factory）を設定します。これらのセッションを通じて、データベースとのすべての対話が行われます。

最後に、declarative_base関数は、モデルクラスが派生するベースクラスを生成します。これらのモデルクラスは、データベースのテーブルを表現します。

これで、FastAPIアプリケーションからデータベースに接続する準備が整いました。次のステップは、CRUD操作を実装することです。これについては次のセクションで説明します。

CRUD操作の基本

CRUDは、データベース操作の基本的な4つの関数を表す頭文字を取ったもので、Create（作成）、Read（読み取り）、Update（更新）、Delete（削除）を指します。

  • Create（作成）: データベースに新しいレコードを作成します。SQLでは、これは通常、INSERTステートメントを使用して行われます。

  • Read（読み取り）: データベースから一つまたは複数のレコードを取得します。これは、SELECTステートメントを使用して行われます。

  • Update（更新）: データベース内の既存のレコードを変更します。これは、UPDATEステートメントを使用して行われます。

  • Delete（削除）: データベースからレコードを削除します。これは、DELETEステートメントを使用して行われます。

これらの操作は、ほとんどのデータベースシステムでサポートされており、データベースとの対話の基本的な構成要素を提供します。

FastAPIとSQLAlchemyを使用すると、これらのCRUD操作をPythonのコードで簡単に実装できます。次のセクションでは、FastAPIを使用したCRUD操作の具体的な実装方法について説明します。

FastAPIでのCRUD操作の実装

FastAPIとSQLAlchemyを使用して、データベースのCRUD操作を実装する方法を説明します。ここでは、前のセクションで作成したSQLiteデータベースと接続します。

まず、データベースのテーブルを表現するモデルクラスを作成します。この例では、Userという名前のテーブルを作成します。

from sqlalchemy import Column, Integer, String

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)

次に、CRUD操作を実装します。これらの操作は、データベースセッションを通じて行われます。

from sqlalchemy.orm import Session
from fastapi import HTTPException

def get_user(db: Session, user_id: int):
    db_user = db.query(User).filter(User.id == user_id).first()
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

def create_user(db: Session, user: User):
    db.add(user)
    db.commit()
    db.refresh(user)
    return user

def update_user(db: Session, user: User):
    db_user = get_user(db, user.id)
    db_user.name = user.name
    db_user.email = user.email
    db_user.hashed_password = user.hashed_password
    db.commit()
    db.refresh(db_user)
    return db_user

def delete_user(db: Session, user_id: int):
    db_user = get_user(db, user_id)
    db.delete(db_user)
    db.commit()
    return {"message": "User deleted"}

最後に、FastAPIのルートを作成して、これらのCRUD操作をWeb APIとして公開します。

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

app = FastAPI()

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

@app.get("/users/{user_id}", response_model=User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    return get_user(db, user_id)

@app.post("/users/", response_model=User)
def create_new_user(user: User, db: Session = Depends(get_db)):
    return create_user(db, user)

@app.put("/users/", response_model=User)
def update_existing_user(user: User, db: Session = Depends(get_db)):
    return update_user(db, user)

@app.delete("/users/{user_id}")
def delete_existing_user(user_id: int, db: Session = Depends(get_db)):
    return delete_user(db, user_id)
以上が、FastAPIとSQLAlchemyを使用したCRUD操作の基本的な実装方法です。これにより、Web APIを通じてデータベースのレコードを操作することができます。ただし、実際のアプリケーションでは、入力の検証、エラーハンドリング、セキュリティ対策など、さらに多くの要素を考慮する必要があります。これらの詳細については、FastAPIの公式ドキュメンテーションを参照してください。

FastAPI公式ドキュメンテーション

エラーハンドリング

エラーハンドリングは、アプリケーションの堅牢性を確保するための重要な要素です。FastAPIでは、エラーハンドリングは非常に簡単に行うことができます。

FastAPIでは、HTTPExceptionを使用してエラーを発生させることができます。HTTPExceptionは、ステータスコードとエラーメッセージを含むHTTPレスポンスを直接返す例外です。

以下に、FastAPIでの基本的なエラーハンドリングの例を示します。

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

上記のコードでは、指定されたitem_idがitemsに存在しない場合、ステータスコード404とともに”Item not found”というエラーメッセージを返します。

また、FastAPIでは、@app.exception_handlerデコレータを使用して、特定の例外に対するカスタムハンドラを定義することも可能です。これにより、特定の例外が発生したときにカスタムレスポンスを返すことができます。

from fastapi import Request
from fastapi.responses import JSONResponse

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"message": f"Oops! {exc.detail}"},
    )

上記のコードでは、HTTPExceptionが発生したときに、カスタムメッセージを含むJSONレスポンスを返します。

これらの方法を使用することで、FastAPIアプリケーションでエラーハンドリングを効果的に行うことができます。ただし、エラーハンドリングはアプリケーションの要件によりますので、適切なエラーハンドリング戦略を選択することが重要です。また、エラーメッセージはユーザーにとって有用で理解しやすいものにすることが重要です。これにより、ユーザーは何が問題であるかを理解し、必要な場合は適切なアクションを取ることができます。

FastAPI公式ドキュメンテーション – エラーハンドリング

テスト方法

FastAPIアプリケーションのテストは、アプリケーションが正しく動作していることを確認するための重要なステップです。FastAPIは、テストを簡単に行うためのツールを提供しています。

FastAPIのテストでは、TestClientを使用します。これは、requestsライブラリと同じように動作しますが、実際にはネットワークを介してリクエストを送信せず、代わりにアプリケーションと直接対話します。これにより、テストが高速で信頼性があり、どこでも実行できます。

以下に、FastAPIでの基本的なテストの例を示します。

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

上記のコードでは、TestClientを使用してアプリケーションのルート（”/”）にGETリクエストを送信し、レスポンスが期待通りであることを確認しています。

FastAPIのテストでは、通常、以下のようなステップを踏みます：

  1. TestClientを作成します。
  2. テストしたいエンドポイントにリクエストを送信します。
  3. レスポンスが期待通りであることを確認します。

これらのステップを踏むことで、FastAPIアプリケーションが正しく動作していることを確認できます。ただし、実際のテストでは、さまざまなパラメータ、異なるリクエスト、エラーケースなど、多くの異なるシナリオをテストする必要があります。

また、FastAPIはPydanticモデルと統合しているため、リクエストとレスポンスのバリデーションも自動的に行われます。これにより、テスト中に予期しないデータが送信されたり、受信されたりするのを防ぐことができます。

FastAPI公式ドキュメンテーション – テスト

まとめ

この記事では、PythonのFastAPIフレームワークを使用してデータベースのCRUD操作を実装する方法について説明しました。FastAPIは、モダンで高速なWebフレームワークであり、Pythonの型ヒントを活用しています。

まず、FastAPIとその特徴について紹介しました。次に、FastAPIを使用してデータベースに接続する方法と、CRUD操作の基本について説明しました。その後、FastAPIとSQLAlchemyを使用してCRUD操作を実装する具体的な方法を示しました。

また、エラーハンドリングについて説明し、FastAPIが提供するHTTPExceptionを使用してエラーを処理する方法を示しました。最後に、FastAPIアプリケーションのテスト方法について説明しました。

FastAPIは、APIの開発を簡単にし、開発者の生産性を向上させる強力なツールです。この記事が、FastAPIを使用したWebアプリケーション開発の一助となることを願っています。引き続き学習を進めて、FastAPIの可能性を最大限に引き出してください。それでは、Happy coding! 🚀

コメントする

メールアドレスが公開されることはありません。 ※ が付いている欄は必須項目です