Contents
FastAPIとPostgreSQLを用いた非同期CRUD実装の概要
FastAPIとPostgreSQLを組み合わせることで、高パフォーマンスな非同期処理が可能になります。特にDatabasesライブラリとasyncpgドライバの組み合わせは、データベースへの接続・クエリ実行に際して、I/O待ち時間を最小限に抑えることが可能です。これにより、多くのユーザーを同時に処理するWebサービスにおいてもスムーズな応答が期待できます。本記事では、こうした技術スタックを用いた非同期CRUD（作成・取得・更新・削除）の実装手順を具体的に解説します。
FastAPIの非同期処理メカニズム
FastAPIはASGIベースで構築されており、非同期処理を実現するための仕組みを備えています。この特徴により、データベース接続や外部API呼び出しなどのI/O待ちが発生する処理を効率的に並列化できます。
ASGIベースの非同期処理フロー
FastAPIはASGI（Asynchronous Server Gateway Interface）に準拠しており、サーバーとアプリケーションの間で非同期通信が可能になります。これにより、リクエストを受けてから応答を返すまでの間に他のタスクが実行され、全体の処理性能が向上します。
async/awaitの基本動作原理
async defで定義された関数は、コルーチンとして扱われ、awaitキーワードで非同期処理（例: データベースクエリ）を待機します。この仕組みにより、I/O待ち中に他のタスクが実行され、全体の処理性能が向上します。
asyncpgの導入と環境構築
asyncpgはPostgreSQL用の非同期Pythonドライバで、FastAPIとの連携において高速なデータベースアクセスを実現します。ここでは、asyncpgのインストール方法と接続設定について解説します。
pip install手順
asyncpgの導入には以下のようにpipコマンドを使用します。
|
1 2 |
pip install asyncpg |
接続文字列の形式
PostgreSQLに接続する際は、以下の形式で接続情報を指定します。ホストやポート番号、ユーザー名、データベース名を適宜変更してください。
|
1 2 |
postgresql+asyncpg://username:password@host:port/database_name |
Databasesライブラリによる非同期DB操作
DatabasesはSQLクエリの実行に特化したライブラリで、FastAPIとasyncpgを連携させる際に有効です。以下では、接続設定からクエリの実行方法までの一連の手順を説明します。
アシンクロナスセッション管理
Databasesは非同期処理に対応しており、connect()メソッドでDBに接続し、execute()やfetch_all()などを使ってクエリを実行できます。以下のコード例は基本的な接続方法です。
|
1 2 3 4 5 |
from databases import Database database = Database("postgresql+asyncpg://user:pass@localhost:5432/mydb") await database.connect() |
クエリビルダの使い方
SQLクエリを実行する際には、query()メソッドでクエリを作成し、fetch_all()やfetch_one()を使って結果を取得します。以下に基本的なSELECTクエリの例を示します。
|
1 2 |
rows = await database.fetch_all("SELECT * FROM users") |
非同期CRUD操作の具体例
FastAPIで非同期処理を活かしたCRUD操作を実装するには、各操作ごとに非同期関数を作成し、asyncpg経由でSQLを実行します。以下にそれぞれの操作のコードサンプルを示します。
create（作成）
新しいレコードをデータベースに挿入します。INSERT INTOクエリを使用し、await database.execute()で実行します。
|
1 2 3 4 |
@app.post("/users") async def create_user(name: str, email: str): await database.execute("INSERT INTO users (name, email) VALUES (:name, :email)", {"name": name, "email": email}) |
read（取得）
特定のユーザー情報を取得します。SELECTクエリを使用し、await database.fetch_one()で結果を取得します。
|
1 2 3 4 5 |
@app.get("/users/{user_id}") async def get_user(user_id: int): user = await database.fetch_one("SELECT * FROM users WHERE id = :id", {"id": user_id}) return user |
update（更新）
既存のユーザー情報を変更します。UPDATEクエリを使用し、await database.execute()で実行します。
|
1 2 3 4 |
@app.put("/users/{user_id}") async def update_user(user_id: int, name: str): await database.execute("UPDATE users SET name = :name WHERE id = :id", {"name": name, "id": user_id}) |
delete（削除）
指定されたユーザーをデータベースから削除します。DELETEクエリを使用し、await database.execute()で実行します。
|
1 2 3 4 |
@app.delete("/users/{user_id}") async def delete_user(user_id: int): await database.execute("DELETE FROM users WHERE id = :id", {"id": user_id}) |
トランザクション制御とエラーハンドリング
複数のクエリを一括で実行する場合、トランザクション管理が必要です。これにより、エラーが発生した際にデータの一貫性が保たれます。
明示的なトランザクション管理
トランザクションはBEGIN, COMMIT, ROLLBACKなどのSQLコマンドで制御できます。以下に複数のクエリを1つのトランザクション内で実行する例を示します。
|
1 2 3 4 |
async with database.transaction(): await database.execute("INSERT INTO users (name, email) VALUES (:name, :email)", {"name": "Alice", "email": "[メールアドレス削除]"}) await database.execute("INSERT INTO posts (title, content, user_id) VALUES (:title, :content, :user_id)", {"title": "Test Post", "content": "Hello World", "user_id": 1}) |
異常時のロールバック処理
エラーが発生した場合、トランザクションをロールバックしてデータの一貫性を保ちます。try-except構文で例外を捕捉し、ROLLBACKを行います。
|
1 2 3 4 5 6 7 |
try: async with database.transaction(): await database.execute("INSERT INTO users (name, email) VALUES (:name, :email)", {"name": "Alice", "email": "[メールアドレス削除]"}) raise ValueError("テストエラー") except Exception as e: print(f"エラー発生: {e}") |
DockerでのPostgreSQL環境構築
ローカル開発向けに、Dockerを使用してPostgreSQLのコンテナを構築します。以下にdocker-compose.ymlの例と起動後の接続方法を示します。
docker-compose.yml例
以下の内容をdocker-compose.ymlファイルとして保存し、docker-compose up -dでサービスを起動してください。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 |
version: '3.8' services: db: image: postgres:15-alpine environment: POSTGRES_USER: user POSTGRES_PASSWORD: pass POSTGRES_DB: mydb ports: - "5432:5432" volumes: - db_data:/var/lib/postgresql/data volumes: db_data: |
コンテナ起動後の接続確認
コンテナが起動したら、PostgreSQLに接続して動作を確認します。以下のコマンドで接続できます。
|
1 2 |
psql -h localhost -U user -d mydb -W |
- 非同期処理によりパフォーマンスの向上が見込まれる
- FastAPIとasyncpg・Databasesを組み合わせた実装方法
- Dockerでローカル環境構築を簡単に実現可能
- トランザクション管理やエラーハンドリングもしっかり対応できる