コンテンツにスキップ

Encode/Databases を用いた非同期SQL（リレーショナル）データベース (非推奨)

情報

これらのドキュメントは近日中に更新予定です。🎉

現在のバージョンは Pydantic v1 を想定しています。

新しいドキュメントには Pydantic v2 が含まれ、Pydantic v2 を使用するように更新されたら、SQLModel が使用されます。

"非推奨"

このチュートリアルは非推奨であり、将来のバージョンで削除されます。

また、encode/databases を FastAPI と共に使用して、async および await を使用してデータベースに接続できます。

互換性があるのは

  • PostgreSQL
  • MySQL
  • SQLite

この例では、SQLiteを使用します。これは、単一のファイルを使用し、Pythonが統合サポートを備えているためです。そのため、この例をコピーしてそのまま実行できます。

後で、本番アプリケーションでは、PostgreSQLのようなデータベースサーバーを使用することをお勧めします。

ヒント

SQLAlchemy ORM に関するセクション(SQL (リレーショナル) データベース)から、データベースで操作を実行するためのユーティリティ関数を使用するなど、FastAPIコードとは独立したアイデアを採用できます。

このセクションでは、Starletteの対応する部分と同等にするために、これらのアイデアを適用していません。

SQLAlchemyのインポートと設定

  • SQLAlchemyをインポートします。
  • metadataオブジェクトを作成します。
  • metadataオブジェクトを使用してテーブルnotesを作成します。
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

ヒント

このコードはすべて純粋なSQLAlchemy Coreであることに注意してください。

databasesはここではまだ何もしていません。

databasesのインポートと設定

  • databasesをインポートします。
  • DATABASE_URLを作成します。
  • databaseオブジェクトを作成します。
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

ヒント

別のデータベース（例えばPostgreSQL）に接続している場合は、DATABASE_URLを変更する必要があります。

テーブルの作成

この例では、同じPythonファイルでテーブルを作成していますが、本番環境では、おそらく移行と統合されたAlembicでテーブルを作成することになります。

ここでは、このセクションはFastAPIアプリケーションを起動する直前に直接実行されます。

  • engineを作成します。
  • metadataオブジェクトからすべてのテーブルを作成します。
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

モデルの作成

以下のためのPydanticモデルを作成します。

  • 作成されるメモ (NoteIn)。
  • 返されるメモ (Note)。
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

これらのPydanticモデルを作成することにより、入力データは検証、シリアライズ (変換)、アノテーション (ドキュメント化) されます。

そのため、インタラクティブなAPIドキュメントで全てを確認できるようになります。

接続と切断

  • FastAPIアプリケーションを作成します。
  • データベースに接続および切断するためのイベントハンドラを作成します。
from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

メモの読み取り

メモを読み取るためのパス操作関数を作成します。

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

注意

awaitを使用してデータベースと通信するため、パス操作関数はasyncで宣言されていることに注意してください。

response_model=List[Note]に注目してください。

これはtyping.Listを使用します。

これにより、出力データがNoteのlistとしてドキュメント化 (および検証、シリアライズ、フィルタリング) されます。

メモの作成

メモを作成するためのパス操作関数を作成します。

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

情報

Pydantic v1では、メソッドは.dict()と呼ばれていましたが、Pydantic v2では非推奨 (ただし、まだサポートされています) となり、.model_dump()に名前が変更されました。

ここの例では、Pydantic v1との互換性のために.dict()を使用していますが、Pydantic v2を使用できる場合は、代わりに.model_dump()を使用する必要があります。

注意

awaitを使用してデータベースと通信するため、パス操作関数はasyncで宣言されていることに注意してください。

{**note.dict(), "id": last_record_id}について

noteはPydanticのNoteオブジェクトです。

note.dict()は、そのデータを持つdictを返します。例えば、以下のようなものです。

{
    "text": "Some note",
    "completed": False,
}

ただし、idフィールドはありません。

そこで、note.dict()からキーと値のペアを含む新しいdictを作成します。

{**note.dict()}

**note.dict()はキーと値のペアを直接「展開」するため、{**note.dict()}は、ほぼnote.dict()のコピーになります。

そして、そのコピーdictを拡張し、別のキーと値のペアを追加します: "id": last_record_id

{**note.dict(), "id": last_record_id}

したがって、返される最終的な結果は、例えば次のようになります。

{
    "id": 1,
    "text": "Some note",
    "completed": False,
}

確認しましょう

このコードをそのままコピーして、http://127.0.0.1:8000/docs でドキュメントを確認できます。

そこですべてのAPIがドキュメント化されており、操作できます。

詳細情報

GitHubページでencode/databasesの詳細を読むことができます。