Python の基本

pgmem パッケージは、pgmem バイナリを同梱したプラットフォーム別の wheel です。実行時の依存はありません。PostgreSQL のドライバはいつも使っているものを入れてください。

インストール

pip

pip install pgmem "psycopg[binary]"

uv

uv add --dev pgmem "psycopg[binary]"

Poetry

poetry add --group dev pgmem "psycopg[binary]"

Python 3.9 以降が必要です。wheel は Linux と macOS の x86-64 と arm64、Windows の x86-64 向けにあります。それ以外のプラットフォームでは go build ./cmd/pgmem でバイナリをビルドし、PGMEM_BINARY でその場所を指定してください。

基本の流れ

1. プロセスを起動します。 pgmem.start() はバイナリを起動し、テンプレートサーバーが接続を受け付けるまで待ちます。

   import pgmem
   
   with pgmem.start(database="app") as pg:
       print(pg.template.dsn)  # postgres://postgres@127.0.0.1:54321/app?sslmode=disable

   ほかに user、postgres -c の設定を並べる params（例：["log_statement=all"]）、サーバーログを標準エラーに流す log=True、バイナリの場所を指定する binary を渡せます。

2. スキーマを登録します。 pg.template.dsn に対してマイグレーションを流します。マイグレーションツールを参照してください。

3. シードデータを入れます。 シードデータを参照してください。

4. 公式ドライバから使います。

   psycopg

   import psycopg
   
   with psycopg.connect(pg.template.dsn) as conn:
       row = conn.execute("SELECT name FROM users WHERE id = %s", (1,)).fetchone()

   asyncpg

   import asyncpg
   
   conn = await asyncpg.connect(pg.template.dsn)
   name = await conn.fetchval("SELECT name FROM users WHERE id = $1", 1)
   await conn.close()

   SQLAlchemy

   from sqlalchemy import create_engine, text
   
   url = pg.template.dsn.replace("postgres://", "postgresql+psycopg://", 1)
   engine = create_engine(url)
   with engine.connect() as conn:
       conn.execute(text("SELECT 1"))

   SQLAlchemy は postgres:// スキームを受け付けないので、方言とドライバを明示します。

マイグレーションツール

生 SQL

from pathlib import Path

with psycopg.connect(pg.template.dsn) as conn:
    conn.execute(Path("schema.sql").read_text())

パラメータがなければ、psycopg は 1 つの文字列に書いた複数の文を受け付けます。

Alembic

from alembic import command
from alembic.config import Config

cfg = Config("alembic.ini")
cfg.set_main_option("sqlalchemy.url", pg.template.dsn.replace("postgres://", "postgresql+psycopg://", 1))
command.upgrade(cfg, "head")

Alembic が生成する env.py は設定から sqlalchemy.url を読むので、このまま動きます。

SQLAlchemy のモデル

from myapp.models import Base

Base.metadata.create_all(create_engine(url))

Django

from django.core.management import call_command

# settings.DATABASES["default"] の HOST、PORT、NAME を pg.template に向ける
call_command("migrate", interactive=False)

ノート

Django 標準のテストランナーは test_<name> というデータベースを作って接続します。pgmem のサーバーは持っているすべてのデータベースを提供するので、この使い方も可能ですが、Django での動作は確認していません。またランナーが実行のたびにマイグレーションを流すことになります。テンプレートにマイグレーションを流し、テストにはフォークの DSN を渡すほうが速く済みます。

シードデータ

生 SQL

with psycopg.connect(pg.template.dsn) as conn:
    conn.execute(Path("seed.sql").read_text())

COPY

rows = [(1, "Frank", "frank@example.com"), (2, "Grace", "grace@example.com")]
with psycopg.connect(pg.template.dsn) as conn, conn.cursor() as cur:
    with cur.copy("COPY users (id, name, email) FROM STDIN") as copy:
        for row in rows:
            copy.write_row(row)

factory_boy

import factory
from sqlalchemy.orm import Session

class UserFactory(factory.alchemy.SQLAlchemyModelFactory):
    class Meta:
        model = User
        sqlalchemy_session_persistence = "commit"

    name = factory.Sequence(lambda n: f"user{n}")
    email = factory.LazyAttribute(lambda u: f"{u.name}@example.com")

with Session(create_engine(url)) as session:
    UserFactory._meta.sqlalchemy_session = session
    UserFactory.create_batch(100)

スナップショットとフォークを直接使う

with pgmem.start(database="app") as pg:
    migrate(pg.template.dsn)
    snap = pg.template.snapshot()        # 開いているトランザクションを待つ。既定は 30 秒
    with snap.fork() as fork:            # 準備済みデータベースの専用の複製
        with psycopg.connect(fork.dsn) as conn:
            conn.execute("DELETE FROM orders")
    with snap.fork() as fork:            # こちらには注文が全部残っている
        ...

snapshot() の前に、テンプレートへの接続はすべてコミットするか閉じてください。次のページの pytest プラグインは、これを全部代わりにやってくれます。