FastAPI x SQLite：簡単なWebアプリ開発例

By /

FastAPI x SQLite：簡単なWebアプリ開発例 ～軽量バックエンドAPI構築の基礎～

はじめに

現代のWebアプリケーション開発において、高速かつ効率的なAPIの構築は不可欠です。Pythonはその汎用性の高さから、バックエンド開発においても非常に人気の高い言語となっています。PythonのWebフレームワークの中でも、FastAPIは近年特に注目を集めています。その最大の特徴は、高いパフォーマンスと、型ヒントに基づいた自動ドキュメント生成機能、そして開発効率の高さです。

一方、データベースとして、リレーショナルデータベース（RDB）は多くのアプリケーションで中心的な役割を果たします。MySQLやPostgreSQLのようなパワフルなRDBも素晴らしいですが、小規模なアプリケーションやプロトタイプ開発、あるいは組み込み用途においては、軽量で設定不要なデータベースが求められることがあります。そこで登場するのがSQLiteです。SQLiteはサーバーを必要とせず、単一のファイルとしてデータベース全体を格納します。学習コストが低く、手軽に始められるのが魅力です。

この記事では、FastAPIを使ってRESTful APIを構築し、そのデータ永続化層としてSQLiteを使用する簡単なWebアプリケーション開発例を詳細に解説します。具体的な例として、シンプルなタスク管理（To-Doリスト）アプリケーションのバックエンドAPIを実装していきます。

このアプリケーションは、以下の機能を持つAPIエンドポイントを提供します。

  • 新しいタスクの作成
  • すべてのタスクの取得
  • 特定のタスクの取得
  • 既存のタスクの更新
  • タスクの削除

対象読者は、PythonでのWebアプリケーション開発に興味がある方、FastAPIの基本的な使い方を知りたい方、軽量なデータベースとしてSQLiteを検討している方です。FastAPIとSQLiteの組み合わせが、どのようにシンプルかつ効果的なバックエンドを構築できるのかを見ていきましょう。

なぜFastAPIとSQLiteなのか？

FastAPIとSQLiteの組み合わせは、いくつかの点で理にかなっています。

  • シンプルさと手軽さ:
    • FastAPIは、開発者がAPIロジックに集中できるよう、最小限の設定で開発を始められます。ASGI（Asynchronous Server Gateway Interface）対応により、非同期処理を容易に記述できます。
    • SQLiteは、データベースサーバーのインストールや設定が不要で、ファイル一つでデータベースを管理できます。これは開発初期段階や、ローカル環境でのテストにおいて非常に便利です。
  • パフォーマンス:
    • FastAPIはStarlette（高性能ASGIフレームワーク）とPydantic（データ検証・シリアライゼーションライブラリ）の上に構築されており、そのパフォーマンスは非常に高い評価を受けています。特に非同期処理を活用することで、I/Oバウンドな処理（データベースアクセスなど）において高いスループットを実現できます。
    • SQLiteは組み込み用途や単一ユーザー/低並行アクセス環境において、十分なパフォーマンスを発揮します。
  • Pythonエコシステム:
    • どちらもPythonの強力なエコシステムの一部です。FastAPIはPydanticやSQLAlchemyといった他の人気ライブラリとの親和性が高く、SQLiteもPythonの標準ライブラリであるsqlite3モジュールでサポートされています。（ただし、非同期処理にはaiosqliteなどの非同期対応ライブラリを使用します）
  • 学習コスト:
    • Pythonに慣れている開発者にとって、FastAPIの学習コストは比較的低いです。型ヒントを積極的に活用するため、IDEの補完機能も強力に働きます。
    • SQLiteは標準的なSQL構文を使用するため、他のRDBを使った経験があれば容易に理解できます。

ただし、SQLiteには限界もあります。特に、高並行な同時書き込みアクセスには向いていません。複数のプロセスやスレッドからの書き込み要求が同時に発生すると、ロックによってパフォーマンスが低下したり、デッドロックが発生したりする可能性があります。本番環境で多数の同時アクセスが予想される場合は、PostgreSQLやMySQLのようなクライアント/サーバー型のRDBを検討すべきでしょう。しかし、学習用途、小規模アプリ、内部ツールなど、同時アクセスが限定的なケースでは、SQLiteは非常に有効な選択肢となります。

この記事では、開発・学習用途を想定し、FastAPIとSQLiteの組み合わせでAPIを構築する手順を詳細に追っていきます。

開発環境の構築

まずは開発に必要な環境を整えましょう。Pythonがインストールされていることを前提とします。推奨バージョンはPython 3.7以上です。

  1. 仮想環境の作成と有効化
    プロジェクトごとに仮想環境を作成するのは良い習慣です。これにより、プロジェクト固有のライブラリの依存関係をシステム全体から分離できます。

    “`bash

    プロジェクトディレクトリを作成

    mkdir fastapi-sqlite-todo
    cd fastapi-sqlite-todo

    仮想環境を作成（’venv’は仮想環境の名前、任意）

    python -m venv venv

    仮想環境を有効化

    Windowsの場合:

    .\venv\Scripts\activate

    macOS/Linuxの場合:

    source venv/bin/activate
    ``
    仮想環境が有効化されると、ターミナルプロンプトの先頭に    (venv)` のような表示が出現します。

  2. 必要なライブラリのインストール
    FastAPIアプリケーションを構築し、SQLiteデータベースを非同期で操作するために、以下のライブラリをインストールします。

    • fastapi: FastAPIフレームワーク本体
    • uvicorn: FastAPIアプリケーションを実行するためのASGIサーバー
    • databases: 非同期処理に対応したデータベースアクセスライブラリ。SQLAlchemy Coreと連携して動作します。
    • aiosqlite: databasesがSQLiteに接続する際に使用する非同期DBドライバー

    bash
    pip install fastapi uvicorn databases aiosqlite
    インストールが完了したら、pip freeze > requirements.txt を実行して、インストールされたライブラリとそのバージョンを記録しておくのが良いでしょう。

データベース設計

タスク管理アプリケーションに必要なデータを考えます。非常にシンプルに、各タスクは「タイトル」「説明」「完了したかどうか」の情報を持つとします。

これをSQLiteデータベースのテーブルとして設計します。テーブル名は tasks としましょう。カラムは以下のように定義します。

  • id: タスクを一意に識別するためのID。整数型で、主キー、自動増分とします。
  • title: タスクのタイトル。文字列型で、必須項目とします。
  • description: タスクの詳細説明。文字列型で、任意項目とします。
  • completed: タスクが完了したかどうかを示す真偽値。SQLiteには真偽値型はありませんが、整数型（1を真、0を偽）で代用するのが一般的です。デフォルト値は偽（0）とします。

このスキーマをコードで表現するために、今回はSQLAlchemy Coreを使用します。SQLAlchemy Coreは、ORM機能（Declarative）を持たない、より低レベルな抽象化レイヤーです。databasesライブラリはSQLAlchemy Coreと連携して動作するため、こちらの利用が適しています。

プロジェクトディレクトリ直下に database.py ファイルを作成し、以下の内容を記述します。

“`python

database.py

import databases
import sqlalchemy

データベース接続URL

SQLiteの場合、’sqlite:///./test.db’ のように指定します。

‘:memory:’ を指定すると、メモリ上に一時的なデータベースを作成します。

永続化する場合はファイルパスを指定します。ここでは ‘./tasks.db’ とします。

DATABASE_URL = “sqlite:///./tasks.db”

databases ライブラリの Database インスタンスを作成

非同期操作はこのインスタンスを介して行います。

database = databases.Database(DATABASE_URL)

SQLAlchemy の MetaData インスタンスを作成

データベースの構造（テーブルやカラムなど）を定義するために使用します。

metadata = sqlalchemy.MetaData()

tasks テーブルの定義

sqlalchemy.Table を使用してテーブルを定義します。

tasks = sqlalchemy.Table(
“tasks”, # テーブル名
metadata, # MetaData インスタンス
# カラム定義
# sqlalchemy.Column(カラム名, データ型, 制約…)
sqlalchemy.Column(“id”, sqlalchemy.Integer, primary_key=True, index=True),
sqlalchemy.Column(“title”, sqlalchemy.String, nullable=False), # NOT NULL 制約
sqlalchemy.Column(“description”, sqlalchemy.String, nullable=True), # NULL 許可
sqlalchemy.Column(“completed”, sqlalchemy.Boolean, default=False) # デフォルト値 False
)

Note on Data Types:

SQLite does not have a built-in Boolean type.

SQLAlchemy maps its Boolean type to an Integer in SQLite by default (1 for True, 0 for False).

This is automatically handled by SQLAlchemy and databases library.

データベースエンジンを作成

これはスキーマ定義のために使用されます。

非同期操作には databases.Database インスタンスを使用し、

こちらのエンジンはテーブル作成のような同期的なメタデータ操作に使用することが多いです。

connect_args={“check_same_thread”: False} は、SQLiteが同一スレッドからのアクセスを

強制するデフォルトの挙動を無効にするための設定です。FastAPIのようなASGIアプリケーションは

複数の非同期タスクを単一スレッドで実行することがあり、この設定がないとエラーになる場合があります。

engine = sqlalchemy.create_engine(
DATABASE_URL, connect_args={“check_same_thread”: False}
)

データベースの初期化（テーブル作成）を行う関数

アプリケーション起動時に一度だけ実行することを想定しています。

def create_database_tables():
# MetaData に定義されたテーブルを engine を使ってデータベースに作成します。
metadata.create_all(engine)

“`

このファイルでは、データベースの接続設定、databases.Database インスタンス、SQLAlchemy Coreを使ったテーブル定義を行っています。また、テーブルを実際にデータベースファイルに作成するための関数 create_database_tables も定義しています。

connect_args={"check_same_thread": False} の設定は、SQLiteとマルチスレッド/非同期環境（特にFastAPIのような）で作業する際の一般的な注意点です。SQLiteはデフォルトで、データベース接続を作成したスレッド以外からの操作を許可しません。FastAPIは内部的に非同期処理やスレッドプールを使用することがあり、この制約に引っかかる可能性があるため、この設定で無効化します。ただし、これは厳密にはスレッドセーフティの問題を完全に解決するわけではない点に注意が必要です。しかし、databasesのような非同期ライブラリが適切に接続プールを管理している限り、多くの場合で問題なく動作します。

FastAPIアプリケーションの基本構造

次に、FastAPIアプリケーションのメインとなるファイルを作成します。プロジェクトディレクトリ直下に main.py ファイルを作成します。

このファイルで、FastAPIアプリケーションインスタンスを作成し、データベース接続の確立・切断をアプリケーションの起動・停止イベントと紐付けます。

“`python

main.py

from fastapi import FastAPI
from database import database, create_database_tables # database.py からインポート

FastAPI アプリケーションインスタンスを作成

app = FastAPI()

アプリケーション起動時のイベントハンドラ

@app.on_event(“startup”) デコレータを使用します。

@app.on_event(“startup”)
async def startup():
print(“Starting up…”)
# データベースに接続
await database.connect()
print(“Database connected.”)
# テーブルが存在しない場合に作成
# NOTE: In a real application, you would use a proper database migration tool
# like Alembic. This is a simplified approach for this example.
create_database_tables()
print(“Database tables checked/created.”)

アプリケーション停止時のイベントハンドラ

@app.on_event(“shutdown”) デコレータを使用します。

@app.on_event(“shutdown”)
async def shutdown():
print(“Shutting down…”)
# データベースから切断
await database.disconnect()
print(“Database disconnected.”)

ルートエンドポイント（オプション）

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to the FastAPI SQLite Todo API!”}

今はこれだけですが、後でタスク関連のルーターを追加します。

from routers import tasks_router

app.include_router(tasks_router, prefix=”/tasks”, tags=[“tasks”])

“`

@app.on_event("startup") でデコレートされた非同期関数は、FastAPIアプリケーションが起動する際に実行されます。ここで database.connect() を呼び出し、データベース接続を確立します。また、create_database_tables() を呼び出して、必要であればテーブルを作成します。（実際のプロダクション環境では、テーブルスキーマの変更を管理するためにAlembicのようなマイグレーションツールを使用するのが一般的ですが、ここではシンプル化のため、起動時に毎回存在チェック＆作成を行います）。

@app.on_event("shutdown") でデコレートされた非同期関数は、アプリケーションが停止する際に実行されます。ここで database.disconnect() を呼び出し、確立したデータベース接続を安全に閉じます。

現時点では、このアプリケーションを実行しても、ルートパス (/) へのアクセスに対して簡単なJSONレスポンスを返すだけです。タスク管理機能はまだ実装されていません。

このアプリケーションを実行するには、ターミナルで以下のコマンドを実行します。

bash
uvicorn main:app --reload

  • main:app: main.py ファイル内の app というFastAPIインスタンスを指定します。
  • --reload: コードの変更を検知して自動的にリロードします。開発中に便利です。

サーバーが起動すると、以下のような出力が表示されるはずです。

INFO: Will watch for changes in these directories: ['/path/to/your/project']
INFO: Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO: Started reloader process [xxxxx]
INFO: Started server process [xxxxx]
INFO: Waiting for application startup.
Starting up...
Database connected.
Database tables checked/created.
INFO: Application startup complete.

ブラウザで http://127.0.0.1:8000 にアクセスすると、{"message": "Welcome to the FastAPI SQLite Todo API!"} というレスポンスが表示されるはずです。ターミナルにはGETリクエストに関するログが表示されます。

Ctrl+C でサーバーを停止すると、Shutting down... と表示され、データベース接続が閉じられるログが出力されます。

Pydanticモデルの定義

FastAPIでは、リクエストボディのバリデーションやレスポンスデータのシリアライゼーションにPydanticを強く推奨しています。データベースのタスクデータに対応するPydanticモデルを定義しましょう。

プロジェクトディレクトリ直下に models.py ファイルを作成します。

“`python

models.py

from pydantic import BaseModel, Field
from typing import Optional

リクエストボディとして使用するモデル (タスク作成/更新用)

class TaskCreate(BaseModel):
# Field(…) は Pydantic >= 1.8 で非推奨になった Field(default=…) の代わりに使われます。
# https://pydantic-docs.helpmanual.io/usage/models/#required-fields
# title は必須項目、最小長さ1文字
title: str = Field(…, min_length=1)
# description は任意項目、デフォルトは None
description: Optional[str] = None
# completed は任意項目、デフォルトは False
completed: bool = False

レスポンスとして返すモデル

データベースから取得したデータに ID を含めるため、TaskCreate を継承します。

class Task(TaskCreate):
# id はデータベースによって生成されるため、レスポンスモデルでのみ定義します。
# 整数型で必須項目です。
id: int

# Pydantic の Config クラスで、ORMモードを有効にします。
# ORMモードを有効にすると、PydanticモデルはORMオブジェクト（この例では SQLAlchemy ResultProxy row）から
# 属性名（例: row.id, row.title）を使ってデータを読み込めるようになります。
# これがないと、辞書のようにキーでアクセス（例: row['id']）する必要があり、不便です。
class Config:
    orm_mode = True # Pydantic v2 では from_attributes=True に変更されます。
    # Pydantic v1.x を使用している場合、 orm_mode = True を使用します。
    # Pydantic v2 を使用している場合、 from_attributes = True を使用します。
    # 互換性のため、両方書いておくことも可能です。
    # from_attributes = True # Pydantic v2 用

Pydantic v2 と v1.x の両方に対応するため、Config クラスは以下のように書くこともできます。

import pydantic

class Task(TaskCreate):

id: int

class Config:

if pydantic.VERSION.startswith(‘2.’):

from_attributes = True

else:

orm_mode = True

“`

  • TaskCreate モデルは、新しいタスクを作成したり、既存のタスクを更新したりする際のリクエストボディの構造を定義します。Pydanticのバリデーション機能により、title が文字列で、かつ空でないことが保証されます。description と completed は任意項目です。
  • Task モデルは、クライアントに返すタスクデータの構造を定義します。TaskCreate を継承し、データベースによって自動生成される id フィールドを追加しています。
  • Config クラス内の orm_mode = True (Pydantic v2では from_attributes = True) は重要です。これにより、データベースから取得したSQLAlchemyの行オブジェクト（属性としてカラム値を持つオブジェクト）を、キー（辞書ライクなアクセス）ではなく属性アクセス（row.id のように）でPydanticモデルに渡せるようになります。FastAPIは内部的にこれを活用して、データベースから取得したデータを簡単にPydanticモデルに変換し、JSONレスポンスとして返すことができます。

データベース操作（CRUD）の実装

データベースに対するCRUD操作（作成、読み取り、更新、削除）を実装するコードを記述します。これらの操作は非同期で行う必要があるため、async および await キーワードを使用します。

プロジェクトディレクトリ直下に crud.py ファイルを作成します。このファイルでは、databases ライブラリと sqlalchemy を組み合わせてデータベース操作を行います。

“`python

crud.py

from databases import Database
from sqlalchemy import select, insert, update, delete

from database import tasks # database.py で定義した tasks テーブルメタデータ
from models import TaskCreate # タスク作成用の Pydantic モデル

タスクを作成する関数

database: databases.Database は依存性注入のために型ヒントとして使用します。

task: TaskCreate はリクエストボディから受け取るデータです。

async def create_task(database: Database, task: TaskCreate):
# SQLAlchemy Core の insert() 関数を使って INSERT クエリを構築
# values() メソッドに Pydantic モデルの dict() を渡すことで、カラム名と値を簡単に設定できます。
query = insert(tasks).values(task.dict())
# execute() メソッドでクエリを実行し、挿入された行のプライマリキーを取得
last_record_id = await database.execute(query)
# 挿入されたタスクを取得して返す
# select() 関数で SELECT クエリを構築
query = select(tasks).where(tasks.c.id == last_record_id)
# fetch_one() メソッドで単一の行を取得
# 結果は sqlalchemy.engine.row.Row オブジェクトになります。
created_task = await database.fetch_one(query)
# Row オブジェクトを Pydantic Task モデルに変換して返す
# models.py で Config.orm_mode = True (or from_attributes=True) を設定しているため、
# 属性アクセスでデータを読み取れます。
return created_task # FastAPIのエンドポイントで Task モデルとして自動的にシリアライズされる

すべてのタスクを取得する関数

async def get_tasks(database: Database):
# SELECT * FROM tasks クエリを構築
query = select(tasks)
# fetch_all() メソッドですべての行を取得
# 結果は sqlalchemy.engine.row.Row オブジェクトのリストになります。
return await database.fetch_all(query) # リストの各要素が Task モデルとして自動的にシリアライズされる

特定のタスクを ID で取得する関数

async def get_task(database: Database, task_id: int):
# SELECT * FROM tasks WHERE id = :task_id クエリを構築
query = select(tasks).where(tasks.c.id == task_id)
# fetch_one() メソッドで単一の行を取得
return await database.fetch_one(query) # 単一の Row オブジェクトが Task モデルとして自動的にシリアライズされる

タスクを更新する関数

task_id: 更新対象のタスクID

task: TaskCreate: 更新データ（title, description, completed を含む）

async def update_task(database: Database, task_id: int, task: TaskCreate):
# UPDATE tasks SET title=:title, description=:description, completed=:completed WHERE id = :task_id クエリを構築
query = update(tasks).where(tasks.c.id == task_id).values(task.dict())
# execute() メソッドでクエリを実行
await database.execute(query)
# 更新後のタスクを取得して返す
# これにより、更新が成功したかどうかと、更新後の最新のタスクデータを確認できます。
return await get_task(database, task_id)

タスクを削除する関数

task_id: 削除対象のタスクID

async def delete_task(database: Database, task_id: int):
# DELETE FROM tasks WHERE id = :task_id クエリを構築
query = delete(tasks).where(tasks.c.id == task_id)
# execute() メソッドでクエリを実行
# 削除された行の数を確認したい場合は、 execute() は行数を返さないため、
# 事前に exists() などで存在確認するか、 DELETE ではなく UPDATE で論理削除を実装するなどの工夫が必要です。
# 今回はシンプルに実行するのみとします。
await database.execute(query)
# 削除成功を示す値を返すか、何も返さない（HTTP 204 No Content など）
# ここでは、削除成功が確認できれば良いので、 True を返すことにします。
# 実際のAPIでは、成功ステータスコード (200 OK or 204 No Content) とメッセージで示すのが一般的です。
return True

“`

この crud.py ファイルには、タスクに関する基本的なデータベース操作を行う非同期関数が定義されています。databases ライブラリは、SQLAlchemy Coreで構築したクエリを受け取り、非同期に実行します。

  • database.execute(query): INSERT, UPDATE, DELETE のようなDML（データ操作言語）クエリや、CREATE TABLEのようなDDL（データ定義言語）クエリを実行します。INSERTの場合は、新しく挿入された行のプライマリキーを返します。
  • database.fetch_one(query): SELECT クエリを実行し、結果セットの最初の1行を取得します。結果はSQLAlchemyの Row オブジェクトとして返されます。
  • database.fetch_all(query): SELECT クエリを実行し、結果セットのすべての行を取得します。結果は Row オブジェクトのリストとして返されます。

これらの関数はデータベースインスタンス（databases.Database型のオブジェクト）を第一引数として受け取ります。これにより、FastAPIの依存性注入を活用して、後でデータベースインスタンスをAPIエンドポイント関数に渡すことができます。

APIエンドポイントの実装

いよいよ、FastAPIのルーターを使ってAPIエンドポイントを定義し、先ほど作成した crud.py の関数を呼び出す部分を実装します。

プロジェクトディレクトリ直下に routers というディレクトリを作成し、その中に tasks.py ファイルを作成します。

“`python

routers/tasks.py

from fastapi import APIRouter, Depends, HTTPException, status
from databases import Database

from database import database # database.py から databases.Database インスタンスをインポート
from crud import create_task, get_tasks, get_task, update_task, delete_task # crud.py から関数をインポート
from models import Task, TaskCreate # Pydantic モデルをインポート

APIRouter インスタンスを作成

このルーターにタスク関連のエンドポイントを登録していきます。

prefix や tags は main.py で include_router する際に指定することもできますが、

ここでデフォルトとして指定することもできます。

tags は OpenAPI ドキュメント（Swagger UI）でのグループ分けに使用されます。

router = APIRouter(
prefix=”/tasks”, # このルーター内の全エンドポイントのパスに “/tasks” が付与されます。
tags=[“tasks”], # Swagger UI でこのルーターのエンドポイントを “tasks” グループにまとめます。
)

依存性注入のための関数

この関数を Depends() と一緒に使うと、FastAPIはリクエスト処理時に database インスタンスを提供します。

async def get_database():

return database # main.py で定義したグローバルな database インスタンスを返すだけ

依存性注入のもう一つの方法は、Depends() に callable (関数など) を直接渡す方法です。

こちらの方がシンプルなので、上記の関数定義は省略し、直接 Depends(database) と記述します。

ただし、この場合 database オブジェクト自体が callable である必要があります。

databases.Database インスタンスは直接 callable ではありません。

したがって、実際には以下のように Callable を返す関数を使う必要があります。

依存性注入のためのデータベースプロバイダ関数

async def get_database() -> Database:
# main.py で初期化されたデータベースインスタンスを返します。
return database

新しいタスクを作成するエンドポイント (POST /tasks/)

response_model=Task: 成功時のレスポンスボディの形式を Task モデルで定義します。

status_code=status.HTTP_201_CREATED: 成功時のHTTPステータスコードを 201 Created に設定します。

@router.post(“/”, response_model=Task, status_code=status.HTTP_201_CREATED)
async def create_task_api(
task: TaskCreate, # リクエストボディは TaskCreate モデルでバリデーションされます。
db: Database = Depends(get_database) # get_database() の戻り値を db 引数に注入します。
):
# crud 関数を呼び出してデータベースにタスクを作成
# crud.create_task は作成したタスクの Row オブジェクトを返します。
# FastAPI はこの Row オブジェクトを response_model=Task に従って自動的に Task モデルに変換し、JSONとして返します。
return await create_task(db, task)

全タスクを取得するエンドポイント (GET /tasks/)

response_model=list[Task]: 成功時のレスポンスボディは Task モデルのリストであることを定義します。

@router.get(“/”, response_model=list[Task])
async def read_tasks_api(db: Database = Depends(get_database)):
# crud 関数を呼び出して全タスクを取得
# crud.get_tasks は Row オブジェクトのリストを返します。
# FastAPI はこのリストの各要素を response_model=list[Task] に従って自動的に Task モデルに変換し、JSONとして返します。
return await get_tasks(db)

特定のタスクを ID で取得するエンドポイント (GET /tasks/{task_id})

task_id: int はパスパラメータとして受け取ります。FastAPIが自動的にint型に変換します。

@router.get(“/{task_id}”, response_model=Task)
async def read_task_api(task_id: int, db: Database = Depends(get_database)):
# crud 関数を呼び出して特定のタスクを取得
task = await get_task(db, task_id)
# タスクが見つからなかった場合は 404 Not Found エラーを返す
if task is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Task not found”)
# 見つかった場合はタスクオブジェクトを返す（自動的に Task モデルにシリアライズされる）
return task

タスクを更新するエンドポイント (PUT /tasks/{task_id})

@router.put(“/{task_id}”, response_model=Task)
async def update_task_api(task_id: int, task: TaskCreate, db: Database = Depends(get_database)):
# 更新対象のタスクがデータベースに存在するか確認
existing_task = await get_task(db, task_id)
if existing_task is None:
# 存在しない場合は 404 Not Found エラーを返す
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Task not found”)

# 存在する場合は crud 関数を呼び出してタスクを更新
# crud.update_task は更新後のタスクの Row オブジェクトを返します。
return await update_task(db, task_id, task)

タスクを削除するエンドポイント (DELETE /tasks/{task_id})

成功時は何も返さない (response_model=None)、ステータスコード 204 No Content が適切ですが、

FastAPI の DELETE デコレーターはデフォルトで 200 を返します。

明示的に status_code=status.HTTP_204_NO_CONTENT を指定することもできますが、

その場合レスポンスボディは空になります。ここでは削除成功を示すメッセージを返すため 200 OK を利用します。

あるいは、response_model=None, status_code=status.HTTP_204_NO_CONTENT とし、return None とします。

シンプルに、削除成功を知らせるメッセージを返す例とします。

@router.delete(“/{task_id}”) # response_modelを指定しない場合、デフォルトはJSON
async def delete_task_api(task_id: int, db: Database = Depends(get_database)):
# 削除対象のタスクがデータベースに存在するか確認
existing_task = await get_task(db, task_id)
if existing_task is None:
# 存在しない場合は 404 Not Found エラーを返す
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=”Task not found”)

# 存在する場合は crud 関数を呼び出してタスクを削除
await delete_task(db, task_id)
# 削除成功を示すレスポンスを返す
# 一般的には、DELETE は成功時に 200 OK または 204 No Content を返します。
# 204 の場合はレスポンスボディは空です。ここでは 200 と成功メッセージを返します。
return {"detail": f"Task with id {task_id} deleted successfully"}

“`

この routers/tasks.py ファイルでは、以下の重要な要素を使用しています。

  • APIRouter: エンドポイントをモジュールごとに整理するためのFastAPIの機能です。関連するエンドポイントを一つのルーターにまとめ、main.py から include_router を使ってアプリケーションに追加します。
  • Depends(): FastAPIの依存性注入システムです。Depends(get_database) と指定することで、各エンドポイント関数はリクエスト処理時に get_database() 関数が返すデータベースインスタンスを受け取ることができます。これにより、データベース接続の管理（接続確立・切断、接続プールの利用など）をFastAPIに任せつつ、クエリ実行部分とエンドポイント定義部分を分離できます。
  • Pydanticモデル (TaskCreate, Task):
    • POST/PUTエンドポイントの引数として task: TaskCreate のように指定することで、FastAPIはリクエストボディを自動的にJSONから TaskCreate モデルに変換し、Pydationによるバリデーションを行います。バリデーションエラーがあれば、自動的に 422 Unprocessable Entity エラーレスポンスを返します。
    • response_model=Task や response_model=list[Task] を指定することで、FastAPIはエンドポイント関数の戻り値を自動的に指定されたPydanticモデルに変換し、JSONレスポンスとしてシリアライズします。前述の Config.orm_mode = True のおかげで、データベースから直接取得したSQLAlchemyの Row オブジェクトをそのまま返すだけで、適切に変換が行われます。
  • HTTPException: エラー発生時にクライアントにHTTPエラーレスポンスを返すために使用します。例えば、指定されたIDのタスクが見つからなかった場合に HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="...") を発生させると、FastAPIは自動的に適切なHTTPレスポンスを生成します。
  • status: fastapi.status モジュールには、一般的なHTTPステータスコードが定数として定義されており、可読性を高めます。

最後に、main.py を更新して、この新しいタスクルーターをアプリケーションに含めます。

“`python

main.py (更新後)

from fastapi import FastAPI
from database import database, create_database_tables
from routers import tasks # routers ディレクトリ内の tasks.py から router インスタンスをインポート

FastAPI アプリケーションインスタンスを作成

app = FastAPI()

アプリケーション起動時のイベントハンドラ

@app.on_event(“startup”)
async def startup():
print(“Starting up…”)
await database.connect()
print(“Database connected.”)
# テーブルが存在しない場合に作成
create_database_tables()
print(“Database tables checked/created.”)

アプリケーション停止時のイベントハンドラ

@app.on_event(“shutdown”)
async def shutdown():
print(“Shutting down…”)
await database.disconnect()
print(“Database disconnected.”)

ルートエンドポイント（オプション）

@app.get(“/”)
async def read_root():
return {“message”: “Welcome to the FastAPI SQLite Todo API!”}

タスクルーターをアプリケーションに含める

prefix=”/tasks” を指定しているため、 tasks.py で定義されたエンドポイントのパスは

例えば “/” は “/tasks” に、 “/{task_id}” は “/tasks/{task_id}” になります。

tags=[“tasks”] は、このルーターに含まれる全てのエンドポイントに “tasks” タグを付けます。

app.include_router(tasks.router, prefix=”/tasks”, tags=[“tasks”])

“`

これで、基本的なCRUD操作を備えたFastAPIアプリケーションが完成しました。

アプリケーションの実行とテスト

アプリケーションを再度起動します。

bash
uvicorn main:app --reload

サーバーが起動したら、APIエンドポイントをテストできます。FastAPIは自動的にOpenAPIドキュメントとSwagger UIを生成してくれるので、それを利用するのが便利です。ブラウザで http://127.0.0.1:8000/docs にアクセスしてください。

Swagger UIが表示され、「tasks」というグループの下に定義したエンドポイントが表示されているはずです。

  • POST /tasks/: 新しいタスクを作成
  • GET /tasks/: 全タスクを取得
  • GET /tasks/{task_id}: 特定のタスクを取得
  • PUT /tasks/{task_id}: 特定のタスクを更新
  • DELETE /tasks/{task_id}: 特定のタスクを削除

Swagger UI上で「Try it out」ボタンをクリックし、パラメータを入力して「Execute」することで、各エンドポイントを簡単に試すことができます。

例えば、新しいタスクを作成するには:

  1. POST /tasks/ のエンドポイントを展開します。
  2. 「Try it out」をクリックします。
  3. Request body に以下のJSONを入力します。
    json
    {
    "title": "Grocery Shopping",
    "description": "Buy milk, eggs, and bread"
    }
  4. 「Execute」をクリックします。
  5. Responses セクションに、作成されたタスクのIDを含むJSONレスポンス（ステータスコード 201 Created）が表示されるはずです。

他のエンドポイントも同様に試してみてください。特に、タスク作成後に返されるIDを使って、GET, PUT, DELETE エンドポイントを試すと、CRUD操作全体を確認できます。

cURLやPostmanでのテスト例

Swagger UIだけでなく、コマンドラインツールである cURL や、APIテストクライアントである Postman / Insomnia などを使ってもAPIをテストできます。

タスク作成 (POST)

bash
curl -X POST http://127.0.0.1:8000/tasks/ \
-H "Content-Type: application/json" \
-d '{"title": "Learn FastAPI", "description": "Read documentation and tutorials"}'
成功すると、作成されたタスクのJSONが返ってきます。

全タスク取得 (GET)

bash
curl http://127.0.0.1:8000/tasks/
タスクのリストを含むJSONが返ってきます。

特定のタスク取得 (GET /tasks/{id})
例えばIDが1のタスクを取得する場合：
bash
curl http://127.0.0.1:8000/tasks/1
IDが1のタスクのJSONが返ってきます。見つからなければ 404 Not Found エラーが返ります。

タスク更新 (PUT /tasks/{id})
例えばIDが1のタスクを完了済みに更新する場合：
bash
curl -X PUT http://127.0.0.1:8000/tasks/1 \
-H "Content-Type: application/json" \
-d '{"title": "Learn FastAPI", "description": "Read documentation and tutorials", "completed": true}'
更新後のタスクのJSONが返ってきます。

タスク削除 (DELETE /tasks/{id})
例えばIDが1のタスクを削除する場合：
bash
curl -X DELETE http://127.0.0.1:8000/tasks/1
削除成功メッセージのJSONが返ってきます。

これらのツールを使って、APIが設計通りに動作することを確認しましょう。

テストコードの記述

APIが正しく機能していることを保証するために、自動化されたテストを書くことは非常に重要です。FastAPIアプリケーションのテストには、pytest と httpx (非同期HTTPクライアント) を使うのが一般的です。

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

bash
pip install pytest httpx pytest-asyncio

  • pytest: テストフレームワーク本体
  • httpx: 非同期処理対応のHTTPクライアント
  • pytest-asyncio: pytestでasyncioを使うためのプラグイン

次に、テストコードを記述します。テスト用のデータベースとして、ファイルではなくメモリ上のSQLiteを使用すると、テストごとにクリーンな状態を簡単に用意できます。

プロジェクトディレクトリ直下に test_main.py ファイルを作成します。（ファイル名は test_*.py または *_test.py である必要があります）。

“`python

test_main.py

import pytest
from httpx import AsyncClient
from sqlalchemy import create_engine

テスト用にメモリ上のデータベースを使用するため、元の設定を上書きする

このテストファイル内でのみ有効な設定変更となります

from database import metadata, tasks, DATABASE_URL as _DATABASE_URL
from main import app

main.py からデータベースインスタンス自体はインポートせず、テスト用のものを使う

テスト用のメモリ上SQLiteデータベースURL

‘sqlite:///:memory:’ はメモリ上DBを意味します。

TEST_DATABASE_URL = “sqlite:///:memory:”

テスト用のデータベースエンジンを作成

テスト用のメタデータ操作（テーブル作成）に使用

check_same_thread=False は開発時と同様に設定

test_engine = create_engine(
TEST_DATABASE_URL, connect_args={“check_same_thread”: False}
)

テストフィクスチャ

各テスト関数の実行前に呼び出され、テスト用のデータベースを準備します。

@pytest.fixture
def setup_database():
# メモリ上データベースにテーブルを作成
metadata.create_all(test_engine)
yield
# テスト終了後、メモリ上DBは自動的に消滅するため、特別なクリーンアップは不要
# ただし、ファイルDBを使う場合はここで削除処理などが必要になります。

テストフィクスチャ

各テスト関数の実行前に呼び出され、テスト用のAsyncClientを提供します。

@pytest.fixture
async def async_client(setup_database): # setup_database フィクスチャに依存
# FastAPIアプリケーションの依存性注入をオーバーライドします。
# production_database_url をテスト用URLに置き換えます。
# FastAPIアプリケーション内で database.py の database インスタンスではなく、
# テスト用のデータベースインスタンスが使われるようにします。
# これは少し複雑ですが、ポイントはFastAPIが依存性注入でデータベースインスタンスを供給する際に、
# テスト環境では差し替えたいということです。

# main.py の database インスタンスを直接インポートして差し替えるのではなく、
# 依存性注入の Depends(get_database) が返す値を差し替えるのがスマートです。
# routers/tasks.py の get_database 関数が返す database インスタンスを
# テスト用の databases.Database インスタンスに差し替えるフィクスチャを作成します。

# テスト用の databases.Database インスタンス
from databases import Database as _Database # 名前衝突を避けるためエイリアス
test_database = _Database(TEST_DATABASE_URL)

# データベース接続を確立
await test_database.connect()

# 依存性注入のオーバーライド関数を定義
async def override_get_database():
    return test_database

# アプリケーションの依存性をオーバーライド
# routers.tasks.get_database にアクセスするために、routers ディレクトリをPythonモジュールとして認識させる必要があります。
# そのためには、routers ディレクトリに __init__.py ファイル（空でよい）を作成してください。
from routers.tasks import get_database # tasks.py から get_database 関数をインポート

app.dependency_overrides[get_database] = override_get_database

# テスト用の AsyncClient を作成
# with構文でクライアントを使用すると、テスト終了後に自動的にクローズされます。
async with AsyncClient(app=app, base_url="http://testserver") as client:
    yield client # クライアントをテスト関数に提供

# テスト終了後、依存性オーバーライドを元に戻す（重要！）
app.dependency_overrides.clear()
# データベース接続を閉じる
await test_database.disconnect()

テスト関数

各テスト関数は async_client フィクスチャを受け取ります。

pytest-asyncio が async テスト関数を正しく実行してくれます。

ルートエンドポイントのテスト

@pytest.mark.asyncio
async def test_read_root(async_client: AsyncClient):
response = await async_client.get(“/”)
assert response.status_code == 200
assert response.json() == {“message”: “Welcome to the FastAPI SQLite Todo API!”}

タスク作成のテスト

@pytest.mark.asyncio
async def test_create_task(async_client: AsyncClient):
task_data = {“title”: “Test Task”, “description”: “This is a test task.”}
response = await async_client.post(“/tasks/”, json=task_data)

assert response.status_code == 201
response_json = response.json()
assert response_json["title"] == task_data["title"]
assert response_json["description"] == task_data["description"]
assert response_json["completed"] is False # デフォルト値の確認
assert "id" in response_json # IDが生成されているか確認
assert isinstance(response_json["id"], int) # IDが整数型か確認

タイトルが空のタスク作成テスト (バリデーションエラー)

@pytest.mark.asyncio
async def test_create_task_empty_title(async_client: AsyncClient):
task_data = {“title”: “”, “description”: “Empty title task.”}
response = await async_client.post(“/tasks/”, json=task_data)

assert response.status_code == 422 # Unprocessable Entity (バリデーションエラー)
# 詳細なエラーレスポンス構造は FastAPI / Pydantic のバージョンに依存する可能性がありますが、
# 通常はエラー内容を示す詳細情報が含まれます。

全タスク取得のテスト

@pytest.mark.asyncio
async def test_read_tasks(async_client: AsyncClient):
# テスト前にタスクを作成しておく
await async_client.post(“/tasks/”, json={“title”: “Task 1”})
await async_client.post(“/tasks/”, json={“title”: “Task 2”})

response = await async_client.get("/tasks/")

assert response.status_code == 200
response_json = response.json()
assert isinstance(response_json, list) # レスポンスがリストか確認
assert len(response_json) >= 2 # 作成したタスクが含まれているか（最低限2つ）

特定タスク取得のテスト (成功ケース)

@pytest.mark.asyncio
async def test_read_single_task(async_client: AsyncClient):
# テスト前にタスクを作成し、そのIDを取得
create_response = await async_client.post(“/tasks/”, json={“title”: “Single Read Test”})
created_task_id = create_response.json()[“id”]

# 作成したタスクのIDを指定して取得
read_response = await async_client.get(f"/tasks/{created_task_id}")

assert read_response.status_code == 200
read_response_json = read_response.json()
assert read_response_json["id"] == created_task_id
assert read_response_json["title"] == "Single Read Test"

特定タスク取得のテスト (NotFound ケース)

@pytest.mark.asyncio
async def test_read_single_task_not_found(async_client: AsyncClient):
# 存在しないであろう大きなIDを指定して取得
response = await async_client.get(“/tasks/9999”)
assert response.status_code == 404
assert response.json() == {“detail”: “Task not found”}

タスク更新のテスト (成功ケース)

@pytest.mark.asyncio
async def test_update_task(async_client: AsyncClient):
# テスト前にタスクを作成し、そのIDを取得
create_response = await async_client.post(“/tasks/”, json={“title”: “Task to Update”})
created_task_id = create_response.json()[“id”]

# 更新データ
update_data = {"title": "Updated Task", "description": "New description", "completed": True}
# 作成したタスクのIDを指定して更新
update_response = await async_client.put(f"/tasks/{created_task_id}", json=update_data)

assert update_response.status_code == 200
update_response_json = update_response.json()
assert update_response_json["id"] == created_task_id
assert update_response_json["title"] == update_data["title"]
assert update_response_json["description"] == update_data["description"]
assert update_response_json["completed"] == update_data["completed"]

タスク更新のテスト (NotFound ケース)

@pytest.mark.asyncio
async def test_update_task_not_found(async_client: AsyncClient):
# 存在しないであろう大きなIDを指定して更新
update_data = {“title”: “Updated Task”, “description”: “New description”, “completed”: True}
response = await async_client.put(“/tasks/9999”, json=update_data)
assert response.status_code == 404
assert response.json() == {“detail”: “Task not found”}

タスク削除のテスト (成功ケース)

@pytest.mark.asyncio
async def test_delete_task(async_client: AsyncClient):
# テスト前にタスクを作成し、そのIDを取得
create_response = await async_client.post(“/tasks/”, json={“title”: “Task to Delete”})
created_task_id = create_response.json()[“id”]

# 作成したタスクのIDを指定して削除
delete_response = await async_client.delete(f"/tasks/{created_task_id}")

assert delete_response.status_code == 200
assert delete_response.json() == {"detail": f"Task with id {created_task_id} deleted successfully"}

# 削除されたことを確認するために、同じIDで取得を試みる
get_response = await async_client.get(f"/tasks/{created_task_id}")
assert get_response.status_code == 404 # 見つからないはず

タスク削除のテスト (NotFound ケース)

@pytest.mark.asyncio
async def test_delete_task_not_found(async_client: AsyncClient):
# 存在しないであろう大きなIDを指定して削除
response = await async_client.delete(“/tasks/9999”)
assert response.status_code == 404
assert response.json() == {“detail”: “Task not found”}

“`

routers ディレクトリに __init__.py ファイルを作成することを忘れないでください。空のファイルで構いません。これにより、Pythonが routers ディレクトリをモジュールとして認識し、from routers.tasks import get_database のようなインポートが可能になります。

bash
touch routers/__init__.py

このテストファイルでは、以下のことを行っています。

  • メモリ上データベース: テストのたびにクリーンな状態にするために、sqlite:///:memory: を使ってメモリ上にデータベースを作成し、テスト終了後に破棄しています。
  • setup_database フィクスチャ: 各テスト関数の実行前に、テスト用のデータベースにテーブルを作成します。
  • async_client フィクスチャ: httpx.AsyncClient を使用して、テスト対象のFastAPIアプリケーション (main.app) に非同期HTTPリクエストを送信するためのクライアントを作成します。また、依存性注入をオーバーライドして、アプリケーションが実際のファイルデータベースではなく、テスト用のメモリ上データベースインスタンスを使用するように設定しています。これは、app.dependency_overrides 属性に、オリジナルの依存性 (routers.tasks.get_database) と、テスト用に差し替えたい関数 (override_get_database) のマッピングを追加することで実現しています。テスト終了後には忘れずにオーバーライドをクリアする必要があります。
  • pytest.mark.asyncio: 非同期テスト関数をpytestに認識させるためのマーカーです。pytest-asyncio プラグインがこれを処理します。
  • テスト関数: 各テスト関数は async_client フィクスチャを受け取り、それを使ってHTTPリクエストを送信し、レスポンスのステータスコードやボディをアサート（検証）します。CRUD操作の各シナリオ（成功、NotFound、バリデーションエラーなど）をテストしています。

テストを実行するには、プロジェクトディレクトリのルートで以下のコマンドを実行します。

bash
pytest

すべてのテストがパスすることを確認してください。

デバッグ

開発中に問題が発生した場合のデバッグ方法について簡単に触れておきます。

  • printデバッグ: 一番シンプルですが効果的な方法です。FastAPIのログは標準出力に出力されるため、print() 関数で変数の中身などを出力することで、処理の流れや値を確認できます。ただし、非同期環境では出力の順序が期待通りにならない場合がある点に注意が必要です。

  • ロギング: より本格的なデバッグにはPythonの logging モジュールを使用します。FastAPIやUvicornも内部的に logging を使用しており、詳細なログレベル（DEBUG, INFO, WARNINGなど）を設定できます。
    “`python
    import logging

    logging.basicConfig(level=logging.DEBUG) # または logging.INFO など
    logger = logging.getLogger(name)

    @router.post(“/”, response_model=Task, status_code=status.HTTP_201_CREATED)
    async def create_task_api(…):
    logger.debug(f”Creating task: {task}”)
    # …
    ``
    * **IDEのデバッガー:** VS CodeやPyCharmのような統合開発環境(IDE)は強力なデバッガーを提供しています。ブレークポイントを設定し、ステップ実行、変数の中身の検査などを行うことができます。Uvicornで実行しているFastAPIアプリケーションにデバッガーをアタッチする方法は、各IDEのドキュメントを参照してください。通常は、    uvicorn main:app –reloadの代わりに、IDEの実行設定でUvicornモジュールを実行するように設定します。
    * **Swagger UI/HTTPクライアント:** APIリクエスト・レスポンスの詳細（ヘッダー、ボディ、ステータスコードなど）を確認するには、Swagger UIやPostman/InsomniaなどのHTTPクライアントが非常に役立ちます。エラーレスポンスの詳細 (    detailフィールドなど) を確認することで、問題の原因を特定しやすくなります。
    * **データベース内容の確認:** SQLiteデータベースの内容を確認するには、DB Browser for SQLite のようなGUIツールを使うと便利です。直接    tasks.db` ファイルを開いて、テーブル構造や挿入されたデータを確認できます。

デプロイに関する考慮事項

開発したFastAPI+SQLiteアプリケーションをインターネット上に公開（デプロイ）する際には、いくつかの考慮事項があります。

  • SQLiteの同時アクセス制限: 前述の通り、SQLiteは同時書き込みアクセスに弱いです。多数のユーザーからのリクエストが同時に発生するようなサービスには向いていません。ユーザー数が限定的な内部ツールや、読み取り主体で書き込みが少ないアプリケーション、あるいは開発・検証環境としての利用が主となります。もし多数の同時アクセスが必要であれば、PostgreSQLやMySQLなどのクライアント/サーバー型RDBへの移行を検討すべきです。
  • データベースファイルの永続化: デプロイ環境によっては、アプリケーションのファイルシステムが一時的なものであったり、頻繁に再起動やデプロイが行われたりします。データベースファイル (tasks.db) が消えてしまわないよう、永続化ストレージ（ボリューム）に配置する必要があります。Dockerを使用する場合は、ボリュームマウントでデータベースファイルをコンテナ外に保存するのが一般的です。
  • 環境変数: データベースの接続URLやその他の設定値は、コードに直書きするのではなく、環境変数から読み込むようにするのがベストプラクティスです。Pythonでは os.environ を使うか、python-dotenv のようなライブラリを使うのが便利です。
  • デプロイ先:
    • VPS (Virtual Private Server): 自分でサーバー環境を構築し、GunicornやUvicorn + Nginx/Caddy のようなリバースプロキシを組み合わせてデプロイできます。柔軟性が高いですが、設定・管理の手間がかかります。
    • PaaS (Platform as a Service): Heroku, Render, Railway など。これらのサービスでは、アプリケーションコードと設定ファイル（DockerfileやProcfile）をアップロードするだけで、ビルド、実行環境の準備、HTTPS対応などを自動で行ってくれます。SQLiteファイルも永続化ボリュームとして提供される場合が多いです（ただし、無料プランなど制限がある場合も）。gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app のように、Gunicornを使って複数のワーカプロセスでFastAPIアプリケーションを実行するのが一般的です。
    • Docker: アプリケーションと依存関係をコンテナにパッケージ化し、Docker Hubなどのレジストリにプッシュします。Dockerに対応した様々な環境（AWS ECS, Google Cloud Run, Kubernetesなど）にデプロイできます。データベースは別のコンテナやマネージドサービスとして用意するのが一般的ですが、小規模なら永続化ボリューム付きのSQLiteコンテナとしてデプロイすることも可能です。

今回の簡単なタスク管理APIであれば、RenderのようなPaaSにDockerイメージとしてデプロイするのが比較的容易です。

まとめと次のステップ

この記事では、FastAPIとSQLiteを組み合わせて、シンプルなタスク管理APIを開発する手順を詳細に解説しました。

  • FastAPIとSQLiteを選択した理由、それぞれの特徴と制限を理解しました。
  • 仮想環境の構築、必要なライブラリのインストールを行いました。
  • SQLAlchemy Coreとdatabasesライブラリを使ったデータベーススキーマ設計とCRUD操作の実装方法を学びました。
  • FastAPIのAPIRouter、Pydanticモデル、依存性注入(Depends)、イベントハンドラ(@app.on_event) を使ってAPIエンドポイントを実装しました。
  • Swagger UIやcURL、Postmanなどを使ってAPIをテストする方法を紹介しました。
  • pytest と httpx を使った自動テストコードの書き方、特にメモリ上データベースと依存性注入のオーバーライドを活用する方法を学びました。
  • デバッグ方法と、デプロイ時の考慮事項（SQLiteの限界、永続化、環境変数、デプロイ先など）について確認しました。

この簡単なタスク管理APIは、FastAPIとSQLiteを使ったバックエンド開発の基礎を示す良い例です。ここからさらに機能を拡張していくことができます。

次のステップ:

  • 認証・認可: APIにアクセスできるユーザーを制限したり、ユーザーごとにタスクを管理したりする機能を追加します。FastAPIでは、JWT (JSON Web Tokens) やOAuth2などの認証方式を簡単に組み込むことができます。
  • ユーザー管理: ユーザー登録、ログイン、プロフィール管理などの機能を追加します。
  • より複雑なデータ構造: タグ機能、期日、優先度など、タスクに関する情報を増やしたり、タスクとプロジェクトのように関連のあるデータ構造を扱ったりします。SQLAlchemy Coreを使ってリレーションシップを扱う方法や、より高レベルなSQLAlchemy ORMの利用を検討します。
  • ページネーションとフィルタリング: タスクが増えてきた場合に、一度にすべてのタスクを返すのではなく、ページごとに取得したり、特定の条件でフィルタリングしたりする機能を追加します。FastAPIの依存性注入を活用して、クエリパラメータを簡単に処理できます。
  • より高度なエラーハンドリング: カスタム例外ハンドラーを実装して、アプリケーション固有のエラーに対するレスポンスを定義します。
  • データベースマイグレーション: Alembicのようなツールを使って、データベーススキーマの変更履歴を管理し、本番環境への安全な適用を可能にします。
  • 別のデータベースへの移行: 同時アクセス性能が必要になった場合、PostgreSQLやMySQLなどのデータベースへの移行を検討します。SQLAlchemy Core/ORMとdatabasesライブラリは多くのRDBに対応しているため、比較的少ないコード変更で移行できる可能性があります。

FastAPIとSQLiteの組み合わせは、小規模プロジェクトや学習用途、プロトタイピングにおいて非常に強力なツールとなります。この記事が、FastAPIを使ったPythonバックエンド開発の第一歩を踏み出す手助けとなれば幸いです。

コメントする

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

上部へスクロール