【チーム開発】FastAPI + PostgreSQL + Docker によるWebAPI開発で詰まった所
DockerでPythonを使用するときに有用な環境変数
ENV PYTHONUNBUFFERED=1
標準出力・標準エラーのストリームのバッファリングを無効に設定する
参考:
Dockerコンテナをずっと起動しておく
tty: true
stdin_open: true
Dockerのitオプションに相当する
参考:
PostgreSQLのhealthcheck
healthcheck:
test: ["CMD-SHELL", "pg_isready -U $${POSTGRES_USER} -d $${POSTGRES_DB} || exit 1"]
interval: 10s
timeout: 10s
retries: 5
start_period: 5s
DBが起動していることをチェックする
参考:
PostgreSQLの設定の変更と構築後のSQL実行
command: -c 'config_file=/etc/postgresql/postgresql.conf'
volumes:
- ./db/script/:/docker-entrypoint-initdb.d/
- ./db/config/postgresql.conf:/etc/postgresql/postgresql.conf
listen_address = '*' # この行は必須
max_connections = 200
参考:
PostgreSQLのコンテナ内ではパスワードが求められない
psql -U postgres -d postgres
パスワードが求められないのは、localhost
に対してはtrust
認証方式が適用されるため。
参考:
Pythonでの環境変数の読み込み
pip install python-dotenv
import os
from dotenv import load_dotenv
load_dotenv()
KEY = os.environ.get("API_KEY")
参考:
SQLModelを使用したモデル設計
class ScriptBase(SQLModel):
title: str = Field(index=True)
content: str
timer: Optional[int] = Field(default=None)
timer_id: Optional[PyObjectId] = Field(default=None)
class Script(ScriptBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
created_at: datetime = Field(default_factory=datetime.now, index=True)
updated_at: datetime = Field(default_factory=datetime.now, sa_column_kwargs={'onupdate': datetime.now}, index=True)
class ScriptPublic(ScriptBase):
id: uuid.UUID
created_at: datetime
updated_at: datetime
class ScriptCreate(ScriptBase):
pass
class ScriptUpdate(ScriptBase):
title: Optional[str] = None
content: Optional[str] = None
timer: Optional[int] = None
timer_id: Optional[PyObjectId] = None
共通する項目をBaseModelとして、継承する
参考:
SQLModelでUUID
import uuid
class Script(ScriptBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
SQLModelでUUIDを使用する場合、default_factory
に関数を渡す。
ここで渡された関数が、DBにアクセスする前に呼ばれる。
参考:
SQLModelでupdated_at
class Script(ScriptBase, table=True):
id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
created_at: datetime = Field(default_factory=datetime.now, index=True)
updated_at: datetime = Field(default_factory=datetime.now, sa_column_kwargs={'onupdate': datetime.now}, index=True)
sa_column_kwargs
に上記のように渡すと良いそう。
ドキュメントに記載は見つけられなかった。
参考:
【参考】階層構造をSQLで設計
今回は階層構造のデータを検索したりすることはなく、階層構造のデータ全体の取得や作成、更新のみを行うため、NoSQLを扱うことにした。
参考:
MongoDBをDockerで構築
mongodb:
image: mongo:7.0.15
container_name: mongodb
ports:
- "27017:27017"
volumes:
- mongodb-store:/data/db
- ./mongodb/configdb:/data/configdb
environment:
- TZ=Asia/Tokyo
env_file:
- ./mongodb/.env
MONGO_INITDB_ROOT_USERNAME=root
MONGO_INITDB_ROOT_PASSWORD=password
MONGO_INITDB_DATABASE=mongo
dockerで構築し、手元で触ってみる
参考:
MongoDBとFastAPIと接続
from motor.motor_asyncio import AsyncIOMotorClient
client = AsyncIOMotorClient(MONGO_URL)
db = client.genkou
PyObjectId = Annotated[str, BeforeValidator(str)]
MongoDBはBSONとしてデータが格納されるが、FastAPIはJSONをエンコード、デコードする。
JSONに直接エンコードできないようなデータに対して、BSONはサポートしており、データが保存される前に文字列に変換する。
参考:
PydanticでMongoDBのモデルを設計
class TimerInterval(BaseModel):
timer: int = Field(gt=0)
intervals: list['TimerInterval'] = []
class TimerBase(BaseModel):
timer_interval: TimerInterval
class TimerCreate(TimerBase):
pass
class TimerPublic(TimerBase):
id: PyObjectId = Field(alias="_id")
model_config = ConfigDict(
populate_by_name=True
)
class TimerUpdate(TimerBase):
timer_interval: Optional[TimerInterval] = None
Field(gt=0)
で正の整数に限定する。
MongoDBでは_idを使用するので、Field(alias="_id")
とpopulate_by_name=True
でエイリアスを設定。
また、Field(...)
はkwargs
で追加の設定をしない限り必要ないとのこと。
参考:
awaitとasync
async def create_timer(timer: TimerCreate):
timer_data = timer.model_dump()
new_timer = await timer_collection.insert_one(timer_data)
created_timer = await timer_collection.find_one({"_id": new_timer.inserted_id})
return created_timer
await
を使用する際は、その外側の関数にasync
が必要。
参考:
マイグレーションツールAlembicを導入
alembic init {マイグレーション環境名}
マイグレーション環境を作成
sqlalchemy.url = postgresql://$(POSTGRES_USER):$(POSTGRES_PASSWORD)@$(POSTGRES_HOST):$(POSTGRES_PORT)/$(POSTGRES_DB)
alembic.ini
のsqlalchemy.url
を編集する。$()
は環境変数名(.envに書くだけではダメ)
from sqlmodel import SQLModel
from genkou.models import Script
###省略###
target_metadata = SQLModel.metadata
###省略###
def run_migrations_online() -> None:
"""Run migrations in 'online' mode.
In this scenario we need to create an Engine
and associate a connection with the context.
"""
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
url = config.get_main_option("sqlalchemy.url") # 追記
with connectable.connect() as connection:
context.configure(
url=url, # 追記
connection=connection, target_metadata=target_metadata
)
with context.begin_transaction():
context.run_migrations()
env.pyを編集
alembic revision --autogenerate -m "create tables"
マイグレーションファイルを自動生成
alembic upgrade head
DBにマイグレーションファイルの内容を反映
参考:
MongoDB Atlasの導入
参考:
Supabaseの導入
参考:
SQLテーブルを変更した際に、マイグレーションができなくなった
FastAPIの方で、SQLModelの定義を変更し、マイグレーションを使用としたところ、エラーが発生。
DBにデータがあることが原因と考え、SupabaseのGUIからテーブルを削除
やり直すが、状況は変わらず。
参考より、downgrade
するも、状況は変わらず、downgrade base
はできるが、revision
はできない。
そこで、revisions/
配下にあるファイルを削除し、やり直したところ、成功。
(原因はわからなかったが、メンターの方によると、テーブルを削除する前に、downgrade
をするといけたかもしれないらしい。)
参考:
Renderでデプロイ
FROM python:3.13.0
# To avoid buffering of stdout and stderr
# ENV PYTHONUNBUFFERED=1
ENV TZ=Asia/Tokyo
WORKDIR /usr/src/app/
COPY ./requirements.txt /usr/src/app/requirements.txt
RUN python -m pip install --no-cache-dir --upgrade -r requirements.txt
COPY ./app/ /usr/src/app/
CMD ["fastapi", "run", "genkou/app.py", "--port", "80"]
開発でdocker-composeを使用する場合でも、デプロイのことを考えると、 ソースファイルのCOPYなどはDockerfileにあらかじめ書いておくと良いと思った
参考:
Pythonのパッケージライブラリのバージョン確認
pip list
--version
で確認することが多いのですが、fastapiではCLIのバージョンが表示されてしまい、確認の方法がわからなかったので、pip list
で確認しました。
参考:
チーム開発におけるGitHubの使い方
同じチームの人から教えてもらいました。
これでタスク管理をすることでとてもスムーズに開発が進んだと思います。
参考: