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

設計／手法／テスト

技術解説

連載 [第21回] :

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

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サーバーのように複数コンテナを使った開発をする際に便利です。

1.
2├── .devcontainer
3│   ├── Dockerfile
4│   ├── devcontainer.json
5│   └── docker-compose.yml
6└── .github
7    └── dependabot.yml

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

開発環境の起動

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

FastAPIのプロジェクトを作成

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

1.
2├── .devcontainer
3├── .github
4├── app
5│   ├── __init__.py
6│   └── main.py
7└── requirements.txt

requirements.txtの作成

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

01annotated-types==0.6.0 ; python_version >= "3.11" and python_version < "4.0"
02anyio==4.3.0 ; python_version >= "3.11" and python_version < "4.0"
03click==8.1.7 ; python_version >= "3.11" and python_version < "4.0"
04colorama==0.4.6 ; python_version >= "3.11" and python_version < "4.0" and (sys_platform == "win32" or platform_system == "Windows")
05fastapi==0.110.2 ; python_version >= "3.11" and python_version < "4.0"
06greenlet==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")
07h11==0.14.0 ; python_version >= "3.11" and python_version < "4.0"
08httptools==0.6.1 ; python_version >= "3.11" and python_version < "4.0"
09idna==3.7 ; python_version >= "3.11" and python_version < "4.0"
10pydantic-core==2.18.1 ; python_version >= "3.11" and python_version < "4.0"
11pydantic==2.7.0 ; python_version >= "3.11" and python_version < "4.0"
12python-dotenv==1.0.1 ; python_version >= "3.11" and python_version < "4.0"
13pyyaml==6.0.1 ; python_version >= "3.11" and python_version < "4.0"
14sniffio==1.3.1 ; python_version >= "3.11" and python_version < "4.0"
15sqlalchemy==2.0.29 ; python_version >= "3.11" and python_version < "4.0"
16starlette==0.37.2 ; python_version >= "3.11" and python_version < "4.0"
17typing-extensions==4.11.0 ; python_version >= "3.11" and python_version < "4.0"
18uvicorn[standard]==0.29.0 ; python_version >= "3.11" and python_version < "4.0"
19uvloop==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"
20watchfiles==0.21.0 ; python_version >= "3.11" and python_version < "4.0"
21websockets==12.0 ; python_version >= "3.11" and python_version < "4.0"

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

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

main.pyの作成

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

01from typing import Union
02 
03from fastapi import FastAPI
04 
05app = FastAPI()
06 
07 
08@app.get("/")
09def read_root():
10    return {"Hello": "World"}
11 
12 
13@app.get("/items/{item_id}")
14def read_item(item_id: int, q: Union[str, None] = None):
15    return {"item_id": item_id, "q": q}

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

`1`	`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データベースチュートリアルの内容を元にしています。

01.
02├── .devcontainer
03├── .github
04├── app
05│   ├── __init__.py
06│   ├── crud.py
07│   ├── database.py
08│   ├── main.py
09│   ├── models.py
10│   └── schemas.py
11└── requirements.txt

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

01from fastapi import Depends, FastAPI, HTTPException
02from sqlalchemy.orm import Session
03 
04from . import crud, models, schemas
05from .database import SessionLocal, engine
06 
07models.Base.metadata.create_all(bind=engine)
08 
09app = FastAPI()
10 
11def get_db():
12    db = SessionLocal()
13    try:
14        yield db
15    finally:
16        db.close()
17 
18 
19@app.post("/users/", response_model=schemas.User)
20def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
21    db_user = crud.get_user_by_email(db, email=user.email)
22    if db_user:
23        raise HTTPException(status_code=400, detail="Email already registered")
24    return crud.create_user(db=db, user=user)
25 
26 
27@app.get("/users/", response_model=list[schemas.User])
28def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
29    users = crud.get_users(db, skip=skip, limit=limit)
30    return users
31 
32 
33@app.get("/users/{user_id}", response_model=schemas.User)
34def read_user(user_id: int, db: Session = Depends(get_db)):
35    db_user = crud.get_user(db, user_id=user_id)
36    if db_user is None:
37        raise HTTPException(status_code=404, detail="User not found")
38    return db_user
39 
40 
41@app.post("/users/{user_id}/items/", response_model=schemas.Item)
42def create_item_for_user(
43    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
44):
45    return crud.create_user_item(db=db, item=item, user_id=user_id)
46 
47 
48@app.get("/items/", response_model=list[schemas.Item])
49def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
50    items = crud.get_items(db, skip=skip, limit=limit)
51    return items

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

01from sqlalchemy.orm import Session
02 
03from . import models, schemas
04 
05 
06def get_user(db: Session, user_id: int):
07    return db.query(models.User).filter(models.User.id == user_id).first()
08 
09 
10def get_user_by_email(db: Session, email: str):
11    return db.query(models.User).filter(models.User.email == email).first()
12 
13 
14def get_users(db: Session, skip: int = 0, limit: int = 100):
15    return db.query(models.User).offset(skip).limit(limit).all()
16 
17 
18def create_user(db: Session, user: schemas.UserCreate):
19    fake_hashed_password = user.password + "notreallyhashed"
20    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
21    db.add(db_user)
22    db.commit()
23    db.refresh(db_user)
24    return db_user
25 
26 
27def get_items(db: Session, skip: int = 0, limit: int = 100):
28    return db.query(models.Item).offset(skip).limit(limit).all()
29 
30 
31def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
32    db_item = models.Item(**item.dict(), owner_id=user_id)
33    db.add(db_item)
34    db.commit()
35    db.refresh(db_item)
36    return db_item

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

01from sqlalchemy import create_engine
02from sqlalchemy.ext.declarative import declarative_base
03from sqlalchemy.orm import sessionmaker
04 
05SQLALCHEMY_DATABASE_URL = "postgresql://postgres:postgres@db/postgres"
06 
07engine = create_engine(SQLALCHEMY_DATABASE_URL)
08SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
09 
10Base = declarative_base()

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

01from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
02from sqlalchemy.orm import relationship
03 
04from .database import Base
05 
06 
07class User(Base):
08    __tablename__ = "users"
09 
10    id = Column(Integer, primary_key=True)
11    email = Column(String, unique=True, index=True)
12    hashed_password = Column(String)
13    is_active = Column(Boolean, default=True)
14 
15    items = relationship("Item", back_populates="owner")
16 
17 
18class Item(Base):
19    __tablename__ = "items"
20 
21    id = Column(Integer, primary_key=True)
22    title = Column(String, index=True)
23    description = Column(String, index=True)
24    owner_id = Column(Integer, ForeignKey("users.id"))
25 
26    owner = relationship("User", back_populates="items")

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

01from pydantic import BaseModel
02 
03 
04class ItemBase(BaseModel):
05    title: str
06    description: str | None = None
07 
08 
09class ItemCreate(ItemBase):
10    pass
11 
12 
13class Item(ItemBase):
14    id: int
15    owner_id: int
16 
17    class Config:
18        from_attributes = True
19 
20 
21class UserBase(BaseModel):
22    email: str
23 
24 
25class UserCreate(UserBase):
26    password: str
27 
28 
29class User(UserBase):
30    id: int
31    is_active: bool
32    items: list[Item] = []
33 
34    class Config:
35        from_attributes = True

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

1・・・
2    environment:
3      POSTGRES_USER: postgres
4      POSTGRES_DB: postgres
5      POSTGRES_PASSWORD: postgres
6・・・

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

`1`	`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ボタンをクリックしてください。

1{
2  "email": "hoge@example.com",
3  "password": "hogehoge"
4}

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

おわりに

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