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

2024年6月14日(金)
石本 達也
実践編第5回の今回は、「Python」と「PostgreSQL」でWebアプリの開発環境を構築し、データベースに接続してデータを読み書きする方法を解説します。

はじめに

第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__.pymain.pyrequirements.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.pydatabase.pymodels.pyschemas.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アプリケーションの開発環境を構築できることが、より実感できたのではないでしょうか。

日本仮想化技術株式会社
Sierやベンチャー企業を経て、現在は日本仮想化技術でDevOps支援サービス「かんたんDevOps」のDev側を担当。「DevOpsを通じて開発者体験を最大化する」をミッションに理想的な開発環境の実現を目指して技術調査や仕組み作りを行っている。

連載バックナンバー

設計/手法/テスト技術解説
第25回

AWSの監視サービス「CloudWatch」でサーバー監視を試してみよう

2024/8/9
本連載も今回で最終回となります。今回は、AWSの監視サービス「CloudWatch」を使って、簡単なサーバー監視を試してみましょう。
設計/手法/テスト技術解説
第24回

CI環境を構築して「ESLint」で静的解析を実行してみよう

2024/7/26
実践編第8回の今回は、「Dev Containers」でCI環境を構築し、静的解析ツール「ESLint」で静的解析を実行するまでの流れを解説します。
設計/手法/テスト技術解説
第23回

テストコードを書いて「GitHub Actions」でCIを実行してみよう

2024/7/12
実践編第7回の今回は、Webフロントエンド開発を例に、テストコードを書いて「GitHub Actions」でCIを実行するまでの流れを解説します。

Think ITメルマガ会員登録受付中

Think ITでは、技術情報が詰まったメールマガジン「Think IT Weekly」の配信サービスを提供しています。メルマガ会員登録を済ませれば、メルマガだけでなく、さまざまな限定特典を入手できるようになります。

Think ITメルマガ会員のサービス内容を見る

他にもこの記事が読まれています