DevOpsを実現するために行うこと・考えること第21回

「Python」＋「PostgreSQL」のWebアプリ環境でデータの読み書きをしてみよう

実践編第5回の今回は、「Python」と「PostgreSQL」でWebアプリの開発環境を構築し、データベースに接続してデータを読み書きする方法を解説します。

2024年6月14日 6:30

はじめに

第19回ではDev Containersの基本を解説しました。今回は、PythonとPostgreSQLを使ったWebアプリケーション開発環境を構築して、データベースに接続してデータの読み書きをしてみましょう。

開発環境の構築

まず、開発環境を構築します。事前に任意のディレクトリからVisual Studio Codeを起動しておきます。Visual Studio Codeが起動したら、コマンドパレット(表示 > コマンドパレット)からadd dev container configuration files...を検索して実行します。

ここでDev Containersの環境を作るためにいくつか質問が表示されますが、次のように回答します。

ワークスペースに構成を追加するを選択
Python 3 & PostgreSQLを選択
3.11-bullseye(既定)を選択。新しいバージョンがある場合は(既定)と書かれているものを選択
機能は何も選択せずにOKを選択

コマンドパレットの実行が完了すると.devcontainerディレクトリが作成されます。このディレクトリには、Dockerfileやdevcontainer.jsonなどの設定ファイルが含まれています。

第19回との違いは、複数のコンテナイメージを取り扱うため、docker-compose.ymlが追加されている点です。ローカル開発環境ではAPIサーバーとDBサーバーのように複数コンテナを使った開発をする際に便利です。

.
├── .devcontainer
│   ├── Dockerfile
│   ├── devcontainer.json
│   └── docker-compose.yml
└── .github
    └── dependabot.yml

ここまでで、開発環境を起動するために必要なファイルは揃いました。次に開発環境を起動してみましょう。

開発環境の起動

コマンドパレットからdev containers: rebuild containerを検索して実行します。下図ように表示されたら、無事に開発環境が起動できています。

FastAPIのプロジェクトを作成

__init__.pyとmain.py、requirements.txtを作成します。__init__.pyは空ファイルにします。

.
├── .devcontainer
├── .github
├── app
│   ├── __init__.py
│   └── main.py
└── requirements.txt

requirements.txtの作成

前回はPoetryツールを使って作成しましたが、今回は事前に作成済みのところからスタートします。以下の内容をrequirements.txtに記述します。

annotated-types==0.6.0 ; python_version >= "3.11" and python_version < "4.0"
anyio==4.3.0 ; python_version >= "3.11" and python_version < "4.0"
click==8.1.7 ; python_version >= "3.11" and python_version < "4.0"
colorama==0.4.6 ; python_version >= "3.11" and python_version < "4.0" and (sys_platform == "win32" or platform_system == "Windows")
fastapi==0.110.2 ; python_version >= "3.11" and python_version < "4.0"
greenlet==3.0.3 ; python_version >= "3.11" and python_version < "4.0" and (platform_machine == "aarch64" or platform_machine == "ppc64le" or platform_machine == "x86_64" or platform_machine == "amd64" or platform_machine == "AMD64" or platform_machine == "win32" or platform_machine == "WIN32")
h11==0.14.0 ; python_version >= "3.11" and python_version < "4.0"
httptools==0.6.1 ; python_version >= "3.11" and python_version < "4.0"
idna==3.7 ; python_version >= "3.11" and python_version < "4.0"
pydantic-core==2.18.1 ; python_version >= "3.11" and python_version < "4.0"
pydantic==2.7.0 ; python_version >= "3.11" and python_version < "4.0"
python-dotenv==1.0.1 ; python_version >= "3.11" and python_version < "4.0"
pyyaml==6.0.1 ; python_version >= "3.11" and python_version < "4.0"
sniffio==1.3.1 ; python_version >= "3.11" and python_version < "4.0"
sqlalchemy==2.0.29 ; python_version >= "3.11" and python_version < "4.0"
starlette==0.37.2 ; python_version >= "3.11" and python_version < "4.0"
typing-extensions==4.11.0 ; python_version >= "3.11" and python_version < "4.0"
uvicorn[standard]==0.29.0 ; python_version >= "3.11" and python_version < "4.0"
uvloop==0.19.0 ; (sys_platform != "win32" and sys_platform != "cygwin") and platform_python_implementation != "PyPy" and python_version >= "3.11" and python_version < "4.0"
watchfiles==0.21.0 ; python_version >= "3.11" and python_version < "4.0"
websockets==12.0 ; python_version >= "3.11" and python_version < "4.0"

ファイルの作成ができたら、次のコマンドを実行してライブラリをインストールします。

pip install --no-cache-dir --upgrade -r requirements.txt

main.pyの作成

FastAPIの公式チュートリアルを元に作成しています。各コードの詳しい説明は公式サイトのドキュメントを参照してください。

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

ファイルが作成できたら、アプリケーションが起動できるか確認してみましょう。

uvicorn app.main:app --host 0.0.0.0 --port 8080

ブラウザでhttp://localhost:8080にアクセスして、{"Hello": "World"}と表示されれば成功です。

データベースの接続

データベースに接続するため、app/ディレクトリにcrud.py、database.py、models.py、schemas.pyを作成します。作成に当たってはFastAPIのSQLデータベースチュートリアルの内容を元にしています。

.
├── .devcontainer
├── .github
├── app
│   ├── __init__.py
│   ├── crud.py
│   ├── database.py
│   ├── main.py
│   ├── models.py
│   └── schemas.py
└── requirements.txt

また、main.pyを以下のように修正します。

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

from . import crud, models, schemas
from .database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()

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


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

crud.pyを以下のように作成します。

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

database.pyを以下のように作成します。

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

SQLALCHEMY_DATABASE_URL = "postgresql://postgres:postgres@db/postgres"

engine = create_engine(SQLALCHEMY_DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

modules.pyを以下のように作成します。

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

schemas.pyを以下のように作成します。

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        from_attributes = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        from_attributes = True

データベース接続する際に必要な情報は.devcontainer/docker-compose.ymlから取得します。今回は、ローカル開発環境で使用することを想定しているため、ユーザー名やパスワードなどは分かりやすいものにしています。外部からアクセス可能なDBを使う場合は、セキュリティに気をつけてください。

・・・
    environment:
      POSTGRES_USER: postgres
      POSTGRES_DB: postgres
      POSTGRES_PASSWORD: postgres
・・・

プログラムの記述が完了したら、アプリケーションを起動して動作確認を行います。

uvicorn app.main:app --host 0.0.0.0 --port 8080

ブラウザでhttp://localhost:8080/docsにアクセスして、FastAPIの自動生成されたドキュメントが表示されれば成功です。Swagger UIからAPIを実行して、データベースにデータが保存されることを確認してみましょう。

defaultセクションにあるPOST /users/のエンドポイントを展開して、Try it outボタンをクリックします。続けてメールアドレスとパスワードを入力してExecuteボタンをクリックしてください。

{
  "email": "hoge@example.com",
  "password": "hogehoge"
}

defaultセクションにあるGET /users/のエンドポイントを展開して、Try it outボタンをクリックします。Executeボタンをクリックすると、データベースに登録されているユーザー情報が取得できます。

おわりに

今回は、PythonとPostgreSQLでWebアプリケーションの開発環境を構築し、データベースに接続してデータの読み書きする方法を解説しました。今回の内容を通じて、Dev Containersを使うことで手軽にWebアプリケーションの開発環境を構築できることが、より実感できたのではないでしょうか。

この記事のキーワード：