はじめに

これは、富士通クラウドテクノロジーズ Advent Calendar 2023 の4日目の記事です。

SQLAlchemy2.0について

SQLAlchemy 2.0が2023年1月にリリースされました。1系からの大きい変更点としてはORMクエリの変更です。

例として社員テーブルを定義しているとすると

src/database/schema.py

from sqlalchemy.orm import Mapped, DeclarativeBase, MappedAsDataclass, mapped_column
from sqlalchemy import String


class Base(MappedAsDataclass, DeclarativeBase):
    """これを継承した subclasses は dataclasses に変換される."""


class Employee(Base):
    """社員テーブルのスキーマを管理するクラス."""

    __tablename__ = "employee"

    id: Mapped[int] = mapped_column(init=False, primary_key=True)
    number: Mapped[str] = mapped_column(String(4))
    name: Mapped[str] = mapped_column(String(24))
    age: Mapped[int]
    birthplace: Mapped[str] = mapped_column(String(24))

1.系ではORMのコードは以下のように書けます。

src/database/employee_old.py

from typing import Optional
from sqlalchemy.orm import Session
from src.database.schema import Employee


class EmployeeTable:
    def __init__(self, *, db: Session) -> None:
        self.db = db

    def select(self, *, number: str) -> Optional[Employee]:
        """idに紐づく従業員情報を取得します."""
        return self.db.query(Employee).filter(Employee.number == number).first()

2.0系のORMのコードは以下のように変更されます。

src/database/employee_new.py

from typing import Optional
from sqlalchemy import select
from sqlalchemy.orm import Session
from src.database.schema import Employee


class EmployeeTable:
    def __init__(self, *, db: Session) -> None:
        self.db = db

    def select(self, *, number: str) -> Optional[Employee]:
        """idに紐づく従業員情報を取得します."""
        return self.db.scalars(select(Employee).filter(Employee.number == number).limit(1)).first()

単体テストについて

DBにアクセスするなど、関数内でデータが閉じていない場合の単体テストはモックを使うことが多いと思います。しかし、今回のようにORMクエリ自体が変更される場合、変更前後でモックを書き換える必要が出てきてしまうため、テストによって実装が担保されているとは言えなくなってしまいます。

統合テストの追加

そこで、実際にDBを立てて接続する統合テストを作成する必要が出てきます。Gitlab CI/CDでは、CI上でDBなどの別コンテナを立てることができるServicesという機能があります。今回はこれを利用して実際にDBに接続するテストを書いていきます。

テストを作成します

test/conftest.py

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.orm.session import close_all_sessions

from main import app
from src.db import get_db
from src.database.schema import Base


class TestingSession(Session):
    def commit(self):
        self.flush()
        self.expire_all()


@pytest.fixture
def test_db():
    engine = create_engine(
        f"mysql+mysqldb://root:password@testdb/employee?charset=utf8mb4",
        pool_pre_ping=True,
    )
    Base.metadata.create_all(bind=engine)

    TestingSessionLocal = sessionmaker(
        class_=TestingSession, autocommit=False, autoflush=False, bind=engine
    )

    db = TestingSessionLocal()

    def override_get_db():
        try:
            yield db
            db.commit()
        except SQLAlchemyError as e:
            assert e is not None
            db.rollback()

    app.dependency_overrides[get_db] = override_get_db

    # テストケース実行
    yield db

    # 後処理
    db.rollback()
    close_all_sessions()
    engine.dispose()


@pytest.fixture()
def test_client() -> TestClient:
    """FastAPIをテスト用のクライアント."""
    return TestClient(app, base_url="http://localhost", raise_server_exceptions=False)

test/test_get_employee.py

from src.database.schema import Employee


class TestGetEmployee:
    def test_success(self, test_db, test_client):
        # DBの準備をする
        employee = Employee(
            number="0101", name="佐藤太郎", age=28, birthplace="北海道札幌市"
        )
        test_db.add(employee)
        test_db.flush()
        test_db.commit()

        get_res = test_client.get("/employee/0101")
        assert get_res.status_code == 200
        response = get_res.json()
        assert response["name"] == "佐藤太郎"
        assert response["age"] == 28
        assert response["birthplace"] == "北海道札幌市"

servicesを使用してDBを作成します。

.gitlab-ci.yml

services:
  - name: mysql:8.0.32
    alias: testdb
    variables:
      TZ: Asia/Tokyo
      MYSQL_ROOT_PASSWORD: password
      MYSQL_DATABASE: employee

test:
  image:
    name: python:3.12-bookworm
  stage: test
  script:
    - python3 -m pip install poetry
    - poetry install --no-root --only main
    - poetry run pytest

テストの実行

1.系のORMクエリの際のテストと2.系のテストの前後で結果に変更がないのを確認します。

$ poetry install --no-root --only main
Creating virtualenv -D9WZ1QFI-py3.12 in /root/.cache/pypoetry/virtualenvs
Installing dependencies from lock file
Package operations: 23 installs, 0 updates, 0 removals
  • Installing idna (3.6)
  • Installing sniffio (1.3.0)
  • Installing typing-extensions (4.8.0)
  • Installing annotated-types (0.6.0)
  • Installing anyio (3.7.1)
  • Installing h11 (0.14.0)
  • Installing pydantic-core (2.14.5)
  • Installing certifi (2023.11.17)
  • Installing click (8.1.7)
  • Installing greenlet (3.0.1)
  • Installing httpcore (1.0.2)
  • Installing packaging (23.2)
  • Installing iniconfig (2.0.0)
  • Installing pluggy (1.3.0)
  • Installing pydantic (2.5.2)
  • Installing starlette (0.27.0)
  • Installing fastapi (0.104.1)
  • Installing httpx (0.25.2)
  • Installing mysqlclient (2.2.0)
  • Installing pytest (7.4.3)
  • Installing ruff (0.1.6)
  • Installing sqlalchemy (2.0.23)
  • Installing uvicorn (0.24.0.post1)
$ poetry run pytest
============================= test session starts ==============================
platform linux -- Python 3.12.0, pytest-7.4.3, pluggy-1.3.0
rootdir: /builds/norikmb/playground
plugins: anyio-3.7.1
collected 1 item
test/test_get_employee.py 
.                                              [100%]
============================== 1 passed in 0.13s ===============================
Cleaning up project directory and file based variables
00:01
Job succeeded

終わりに

明日は @Authns さんの『samber/loを使ってABC329のA-Fを解いてみた』です。競プロ勢の解説楽しみですね！