SQLAlchemy ও Alembic দিয়ে ডেটাবেস

docker run -d --name bitbyte-pg -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:18

mkdir -p ~/projects/prompt-db/app ~/projects/prompt-db/tests
cd ~/projects/prompt-db
uv init --bare -q
uv add -q "fastapi[standard]" sqlalchemy alembic "psycopg[binary]"
uv add -q --dev pytest httpx2
# with pip: python -m venv .venv && source .venv/bin/activate && pip install -q "fastapi[standard]" sqlalchemy alembic "psycopg[binary]" pytest httpx2
touch app/__init__.py
cat >> pyproject.toml <<'EOF'

[tool.fastapi]
entrypoint = "app.main:app"
EOF
export DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/p10_prompts
echo ok

ok

import psycopg

with psycopg.connect("postgresql://postgres:postgres@localhost:5432/postgres", autocommit=True) as conn:
    for name in ("p10_prompts", "p10_prompts_test"):
        conn.execute(f"DROP DATABASE IF EXISTS {name} WITH (FORCE)")
        conn.execute(f"CREATE DATABASE {name}")
    rows = conn.execute("SELECT datname FROM pg_database WHERE datname LIKE 'p10%' ORDER BY 1")
    print([row[0] for row in rows])

['p10_prompts', 'p10_prompts_test']

# file: app/db.py
"""Engine, session factory and the declarative base."""
import os
from collections.abc import Iterator

from sqlalchemy import MetaData, create_engine
from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker

DATABASE_URL = os.environ["DATABASE_URL"]      # no default: a missing setting should fail loudly

# Named constraints make migrations predictable and let us map errors to messages.
NAMING = {
    "ix": "ix_%(column_0_label)s",
    "uq": "uq_%(table_name)s_%(column_0_N_name)s",
    "ck": "ck_%(table_name)s_%(constraint_name)s",
    "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
    "pk": "pk_%(table_name)s",
}


class Base(DeclarativeBase):
    metadata = MetaData(naming_convention=NAMING)


engine = create_engine(DATABASE_URL, pool_size=5, max_overflow=5, pool_pre_ping=True)
SessionLocal = sessionmaker(engine, expire_on_commit=False)


def get_session() -> Iterator[Session]:
    """FastAPI dependency: one session per request, always closed, rolled back if not committed."""
    with SessionLocal() as session:
        yield session

# file: app/models.py
"""Tables: prompts (one row per name) and prompt_versions (one row per saved version)."""
from datetime import datetime

from sqlalchemy import CheckConstraint, DateTime, ForeignKey, String, Text, UniqueConstraint, func
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.db import Base


class Prompt(Base):
    __tablename__ = "prompts"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(60), unique=True)
    created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())

    versions: Mapped[list["PromptVersion"]] = relationship(
        back_populates="prompt", order_by="PromptVersion.version",
        cascade="all, delete-orphan", passive_deletes=True,
    )


class PromptVersion(Base):
    __tablename__ = "prompt_versions"
    __table_args__ = (
        UniqueConstraint("prompt_id", "version"),
        CheckConstraint("temperature BETWEEN 0 AND 2", name="temperature_range"),
    )

    id: Mapped[int] = mapped_column(primary_key=True)
    prompt_id: Mapped[int] = mapped_column(ForeignKey("prompts.id", ondelete="CASCADE"), index=True)
    version: Mapped[int]
    template: Mapped[str] = mapped_column(Text)
    temperature: Mapped[float] = mapped_column(server_default="0.2")

    prompt: Mapped[Prompt] = relationship(back_populates="versions")

uv run alembic init migrations > /dev/null
ls
ls migrations

alembic.ini
app
migrations
pyproject.toml
tests
uv.lock
README
env.py
script.py.mako
versions

# file: migrations/env.py
from logging.config import fileConfig

from alembic import context
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool

from app import models  # noqa: F401  (importing registers the tables on Base.metadata)
from app.db import DATABASE_URL, Base

config = context.config
fileConfig(config.config_file_name, disable_existing_loggers=False)

engine = create_engine(DATABASE_URL, poolclass=NullPool)
with engine.connect() as connection:
    context.configure(connection=connection, target_metadata=Base.metadata, compare_type=True)
    with context.begin_transaction():
        context.run_migrations()

export QUIET='Context impl|transactional DDL|autogenerate plugin|Detected sequence'
uv run alembic revision --autogenerate -m "create prompts and versions" --rev-id 0001 2>&1 | grep -vE "$QUIET"
sed -n '/^def upgrade/,/^def downgrade/{/^def downgrade/!p}' migrations/versions/0001_*.py

INFO  [alembic.autogenerate.compare.tables] Detected added table 'prompts'
INFO  [alembic.autogenerate.compare.tables] Detected added table 'prompt_versions'
INFO  [alembic.autogenerate.compare.constraints] Detected added index 'ix_prompt_versions_prompt_id' on '('prompt_id',)'
Generating /home/nadia/projects/prompt-db/migrations/versions/0001_create_prompts_and_versions.py ...  done
def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('prompts',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('name', sa.String(length=60), nullable=False),
    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_prompts')),
    sa.UniqueConstraint('name', name=op.f('uq_prompts_name'))
    )
    op.create_table('prompt_versions',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('prompt_id', sa.Integer(), nullable=False),
    sa.Column('version', sa.Integer(), nullable=False),
    sa.Column('template', sa.Text(), nullable=False),
    sa.Column('temperature', sa.Double(), server_default='0.2', nullable=False),
    sa.CheckConstraint('temperature BETWEEN 0 AND 2', name=op.f('ck_prompt_versions_temperature_range')),
    sa.ForeignKeyConstraint(['prompt_id'], ['prompts.id'], name=op.f('fk_prompt_versions_prompt_id_prompts'), ondelete='CASCADE'),
    sa.PrimaryKeyConstraint('id', name=op.f('pk_prompt_versions')),
    sa.UniqueConstraint('prompt_id', 'version', name=op.f('uq_prompt_versions_prompt_id_version'))
    )
    op.create_index(op.f('ix_prompt_versions_prompt_id'), 'prompt_versions', ['prompt_id'], unique=False)
    # ### end Alembic commands ###

uv run alembic upgrade head 2>&1 | grep -vE "$QUIET"
uv run alembic current 2>&1 | grep -vE "$QUIET"

INFO  [alembic.runtime.migration] Running upgrade  -> 0001, create prompts and versions
0001 (head)

sed -i '/created_at: Mapped/a\    owner: Mapped[str] = mapped_column(String(60), server_default="team")' app/models.py
uv run alembic revision --autogenerate -m "add prompt owner" --rev-id 0002 2>&1 | grep -vE "$QUIET"
grep -h 'op\.' migrations/versions/0002_*.py

INFO  [alembic.autogenerate.compare.tables] Detected added column 'prompts.owner'
Generating /home/nadia/projects/prompt-db/migrations/versions/0002_add_prompt_owner.py ...  done
    op.add_column('prompts', sa.Column('owner', sa.String(length=60), server_default='team', nullable=False))
    op.drop_column('prompts', 'owner')

uv run alembic upgrade head 2>&1 | grep -vE "$QUIET"
uv run alembic downgrade -1 2>&1 | grep -vE "$QUIET"
uv run alembic current 2>&1 | grep -vE "$QUIET"
uv run alembic upgrade head 2>&1 | grep -vE "$QUIET"
uv run alembic history 2>&1 | grep -vE "$QUIET"

INFO  [alembic.runtime.migration] Running upgrade 0001 -> 0002, add prompt owner
INFO  [alembic.runtime.migration] Running downgrade 0002 -> 0001, add prompt owner
0001
INFO  [alembic.runtime.migration] Running upgrade 0001 -> 0002, add prompt owner
0001 -> 0002 (head), add prompt owner
<base> -> 0001, create prompts and versions

# file: app/schemas.py
from pydantic import BaseModel, ConfigDict, Field

from app.models import Prompt


class PromptIn(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    name: str = Field(min_length=3, max_length=60, pattern=r"^[a-z0-9-]+$")
    template: str = Field(min_length=1, max_length=4000)
    temperature: float = Field(default=0.2, ge=0, le=2)


class PromptOut(BaseModel):
    id: int
    name: str
    version: int
    template: str
    temperature: float

    @classmethod
    def from_prompt(cls, prompt: Prompt) -> "PromptOut":
        latest = prompt.versions[-1]
        return cls(id=prompt.id, name=prompt.name, version=latest.version,
                   template=latest.template, temperature=latest.temperature)


class PromptPage(BaseModel):
    items: list[PromptOut]
    next_cursor: int | None

# file: app/repository.py
from collections.abc import Sequence

from sqlalchemy import select
from sqlalchemy.orm import Session, selectinload

from app.models import Prompt, PromptVersion


class PromptRepository:
    def __init__(self, session: Session) -> None:
        self.session = session

    def get(self, prompt_id: int) -> Prompt | None:
        return self.session.get(Prompt, prompt_id)

    def add(self, name: str, template: str, temperature: float) -> Prompt:
        prompt = Prompt(name=name, versions=[PromptVersion(version=1, template=template, temperature=temperature)])
        self.session.add(prompt)
        self.session.flush()
        return prompt

    def add_version(self, prompt: Prompt, template: str, temperature: float) -> PromptVersion:
        version = PromptVersion(version=prompt.versions[-1].version + 1, template=template, temperature=temperature)
        prompt.versions.append(version)
        self.session.flush()
        return version

    def list(self, *, q: str | None = None, after_id: int = 0, limit: int = 20) -> Sequence[Prompt]:
        stmt = (select(Prompt).options(selectinload(Prompt.versions))
                .where(Prompt.id > after_id).order_by(Prompt.id).limit(limit))
        if q:
            stmt = stmt.where(Prompt.name.icontains(q, autoescape=True))
        return self.session.scalars(stmt).all()

# file: app/service.py
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.models import Prompt
from app.repository import PromptRepository
from app.schemas import PromptIn

class AppError(Exception):
    status_code = 400
    code = "bad_request"


class NotFound(AppError):
    status_code = 404
    code = "not_found"


class Conflict(AppError):
    status_code = 409
    code = "conflict"


CONFLICTS = {
    "uq_prompts_name": "A prompt with this name already exists",
    "uq_prompt_versions_prompt_id_version": "Another request saved a new version first; reload and try again",
}


def conflict_message(exc: IntegrityError) -> str:
    constraint = getattr(getattr(exc.orig, "diag", None), "constraint_name", None)
    return CONFLICTS.get(constraint, "The change conflicts with existing data")


class PromptService:
    def __init__(self, session: Session) -> None:
        self.session = session
        self.repo = PromptRepository(session)

    def get(self, prompt_id: int) -> Prompt:
        prompt = self.repo.get(prompt_id)
        if prompt is None:
            raise NotFound(f"Prompt {prompt_id} not found")
        return prompt

    def create(self, data: PromptIn) -> Prompt:
        try:
            prompt = self.repo.add(data.name, data.template, data.temperature)
            self.session.commit()
        except IntegrityError as exc:
            self.session.rollback()
            raise Conflict(conflict_message(exc)) from exc
        return prompt

    def new_version(self, prompt_id: int, data: PromptIn) -> Prompt:
        prompt = self.get(prompt_id)
        if data.name != prompt.name:
            raise Conflict("The name of a prompt cannot change; create a new prompt instead")
        try:
            self.repo.add_version(prompt, data.template, data.temperature)
            self.session.commit()
        except IntegrityError as exc:
            self.session.rollback()
            raise Conflict(conflict_message(exc)) from exc
        return prompt

    def list(self, *, q: str | None, cursor: int, limit: int) -> tuple[list[Prompt], int | None]:
        rows = list(self.repo.list(q=q, after_id=cursor, limit=limit + 1))
        page = rows[:limit]
        return page, (page[-1].id if len(rows) > limit else None)

# file: app/main.py
from typing import Annotated

from fastapi import Depends, FastAPI, Path, Query, Request, status
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session

from app.db import get_session
from app.schemas import PromptIn, PromptOut, PromptPage
from app.service import AppError, PromptService

SessionDep = Annotated[Session, Depends(get_session)]


def get_service(session: SessionDep) -> PromptService:
    return PromptService(session)


ServiceDep = Annotated[PromptService, Depends(get_service)]
PromptId = Annotated[int, Path(ge=1)]


def create_app() -> FastAPI:
    app = FastAPI(title="BitByte Prompt Library (database edition)")

    @app.exception_handler(AppError)
    async def handle_app_error(request: Request, exc: AppError) -> JSONResponse:
        body = {"error": {"code": exc.code, "message": str(exc), "details": None}}
        return JSONResponse(body, status_code=exc.status_code)

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/v1/prompts")
    def list_prompts(
        service: ServiceDep,
        q: Annotated[str | None, Query(max_length=60, description="Name contains this text")] = None,
        cursor: Annotated[int, Query(ge=0, description="next_cursor from the previous page")] = 0,
        limit: Annotated[int, Query(ge=1, le=100)] = 20,
    ) -> PromptPage:
        rows, next_cursor = service.list(q=q, cursor=cursor, limit=limit)
        return PromptPage(items=[PromptOut.from_prompt(row) for row in rows], next_cursor=next_cursor)

    @app.post("/v1/prompts", status_code=status.HTTP_201_CREATED)
    def create_prompt(data: PromptIn, service: ServiceDep) -> PromptOut:
        return PromptOut.from_prompt(service.create(data))

    @app.get("/v1/prompts/{prompt_id}")
    def get_prompt(prompt_id: PromptId, service: ServiceDep) -> PromptOut:
        return PromptOut.from_prompt(service.get(prompt_id))

    @app.put("/v1/prompts/{prompt_id}", summary="Save a new version of a prompt")
    def update_prompt(prompt_id: PromptId, data: PromptIn, service: ServiceDep) -> PromptOut:
        return PromptOut.from_prompt(service.new_version(prompt_id, data))

    return app


app = create_app()

from sqlalchemy import func, select

from app.db import SessionLocal
from app.models import Prompt, PromptVersion


def count(session):
    return session.scalar(select(func.count()).select_from(Prompt))


with SessionLocal() as session, SessionLocal() as observer:
    session.add(Prompt(name="draft", versions=[PromptVersion(version=1, template="Draft: {input}")]))
    session.flush()                                   # INSERT is sent, the transaction is still open
    print("writer sees:", count(session), "| observer sees:", count(observer))
    session.rollback()
    print("after rollback:", count(session))

writer sees: 1 | observer sees: 0
after rollback: 0

from sqlalchemy.exc import IntegrityError

with SessionLocal() as session:
    try:
        session.add(Prompt(name="atomic", versions=[
            PromptVersion(version=1, template="fine {input}", temperature=0.2),
            PromptVersion(version=2, template="broken {input}", temperature=9),
        ]))
        session.commit()
    except IntegrityError as exc:
        session.rollback()
        print(type(exc.orig).__name__, "-", exc.orig.diag.constraint_name)
    print("prompts saved:", count(session))

CheckViolation - ck_prompt_versions_temperature_range
prompts saved: 0

with SessionLocal() as session:
    first = Prompt(name="support-answer", versions=[PromptVersion(version=1, template="Answer: {input}")])
    second = Prompt(name="summarise", versions=[PromptVersion(version=1, template="Summarise: {input}")])
    session.add_all([first, second])
    session.commit()
    prompt_id = first.id

with SessionLocal() as session:
    session.add(Prompt(name="support-answer"))
    try:
        session.commit()
    except IntegrityError as exc:
        session.rollback()
        print("duplicate name:", exc.orig.sqlstate, exc.orig.diag.constraint_name)

with SessionLocal() as a, SessionLocal() as b:
    next_a = len(a.get(Prompt, prompt_id).versions) + 1       # both writers read "version 1 is the latest"
    next_b = len(b.get(Prompt, prompt_id).versions) + 1
    a.add(PromptVersion(prompt_id=prompt_id, version=next_a, template="A saves first {input}"))
    a.commit()
    b.add(PromptVersion(prompt_id=prompt_id, version=next_b, template="B arrives late {input}"))
    try:
        b.commit()
    except IntegrityError as exc:
        b.rollback()
        print("second writer:", exc.orig.sqlstate, exc.orig.diag.constraint_name)

duplicate name: 23505 uq_prompts_name
second writer: 23505 uq_prompt_versions_prompt_id_version

from sqlalchemy import event
from sqlalchemy.orm import selectinload

from app.db import engine

statements = []


@event.listens_for(engine, "before_cursor_execute")
def record(conn, cursor, statement, parameters, context, executemany):
    statements.append(statement)


def load(*options):
    statements.clear()
    with SessionLocal() as session:
        prompts = session.scalars(select(Prompt).options(*options)).all()
        versions = sum(len(prompt.versions) for prompt in prompts)
    return f"{len(prompts)} prompts, {versions} versions, {len(statements)} statements"


print("lazy:        ", load())
print("selectinload:", load(selectinload(Prompt.versions)))

lazy:         2 prompts, 3 versions, 3 statements
selectinload: 2 prompts, 3 versions, 2 statements

uv run fastapi run --port 8000 > server.log 2>&1 &
export SERVER_PID=$!
curl -s --retry 20 --retry-connrefused --retry-delay 0 http://localhost:8000/health

{"status":"ok"}

curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts -H 'Content-Type: application/json' \
  -d '{"name": "classify-ticket", "template": "Classify this ticket: {input}", "temperature": 0}'
curl -s -w ' %{http_code}\n' -X POST http://localhost:8000/v1/prompts -H 'Content-Type: application/json' \
  -d '{"name": "summarise", "template": "Another summariser"}'
curl -s -o /dev/null -w '%{http_code}\n' -X POST http://localhost:8000/v1/prompts -H 'Content-Type: application/json' \
  -d '{"name": "bad-temp", "template": "x", "temperature": 5}'
curl -s -w ' %{http_code}\n' -X PUT http://localhost:8000/v1/prompts/3 -H 'Content-Type: application/json' \
  -d '{"name": "support-answer", "template": "Answer politely: {input}"}'
curl -s -w ' %{http_code}\n' http://localhost:8000/v1/prompts/999

{"id":6,"name":"classify-ticket","version":1,"template":"Classify this ticket: {input}","temperature":0.0} 201
{"error":{"code":"conflict","message":"A prompt with this name already exists","details":null}} 409
422
{"id":3,"name":"support-answer","version":3,"template":"Answer politely: {input}","temperature":0.2} 200
{"error":{"code":"not_found","message":"Prompt 999 not found","details":null}} 404

curl -s 'http://localhost:8000/v1/prompts?limit=2' | python -m json.tool | grep -E '"(name|version|next_cursor)"'
echo ---
curl -s 'http://localhost:8000/v1/prompts?limit=2&cursor=4' | python -m json.tool | grep -E '"(name|version|next_cursor)"'
echo ---
curl -s 'http://localhost:8000/v1/prompts?q=SUM' | python -c "import sys, json; print([p['name'] for p in json.load(sys.stdin)['items']])"

            "name": "support-answer",
            "version": 3,
            "name": "summarise",
            "version": 1,
    "next_cursor": 4
---
            "name": "classify-ticket",
            "version": 1,
    "next_cursor": null
---
['summarise']

kill "$SERVER_PID"

from sqlalchemy import create_engine, text
from sqlalchemy.exc import TimeoutError as PoolTimeout

from app.db import DATABASE_URL

small = create_engine(DATABASE_URL, pool_size=2, max_overflow=0, pool_timeout=1)
first, second = small.connect(), small.connect()
print(small.pool.status())
try:
    small.connect()
except PoolTimeout as exc:
    print(str(exc).split(" (")[0])
first.close()
with small.connect() as connection:               # a connection was returned, so this works
    print(connection.execute(text("SELECT 'served'")).scalar())
second.close()
small.dispose()

Pool size: 2  Connections in pool: 0 Current Overflow: 0 Current Checked out connections: 2
QueuePool limit of size 2 overflow 0 reached, connection timed out, timeout 1.00
served

import asyncio

from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


async def main() -> None:
    async_engine = create_async_engine(DATABASE_URL)
    async_session = async_sessionmaker(async_engine, expire_on_commit=False)
    async with async_session() as session:
        stmt = select(Prompt).options(selectinload(Prompt.versions)).order_by(Prompt.id)
        prompts = (await session.scalars(stmt)).all()
        print([(prompt.name, len(prompt.versions)) for prompt in prompts])
    await async_engine.dispose()


asyncio.run(main())

[('support-answer', 3), ('summarise', 1), ('classify-ticket', 1)]

# file: tests/conftest.py
import os

import pytest
from alembic import command
from alembic.config import Config
from fastapi.testclient import TestClient
from sqlalchemy import text

from app.db import DATABASE_URL, engine
from app.main import create_app

assert DATABASE_URL.endswith("_test"), "refusing to run tests against a non-test database"


@pytest.fixture(scope="session", autouse=True)
def schema():
    command.upgrade(Config("alembic.ini"), "head")     # the real migrations build the test schema
    yield
    engine.dispose()


@pytest.fixture(autouse=True)
def empty_tables():
    yield
    with engine.begin() as connection:
        connection.execute(text("TRUNCATE prompts RESTART IDENTITY CASCADE"))


@pytest.fixture
def client():
    with TestClient(create_app()) as test_client:
        yield test_client

# file: tests/test_prompts.py
BODY = {"name": "support-answer", "template": "Answer: {input}"}


def test_duplicate_name_is_a_409(client):
    assert client.post("/v1/prompts", json=BODY).status_code == 201
    duplicate = client.post("/v1/prompts", json=BODY)
    assert duplicate.status_code == 409
    assert duplicate.json()["error"]["code"] == "conflict"


def test_cursor_pagination_and_clean_state(client):
    for number in range(3):
        client.post("/v1/prompts", json={**BODY, "name": f"prompt-{number}"})
    page = client.get("/v1/prompts?limit=2").json()
    assert [item["id"] for item in page["items"]] == [1, 2]      # ids restart: the tables were emptied
    assert page["next_cursor"] == 2
    assert client.get("/v1/prompts?limit=2&cursor=2").json()["next_cursor"] is None

DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/p10_prompts_test uv run python -m pytest -q 2>&1 | tail -3

..                                                                                           [100%]
2 passed in 0.21s

from sqlalchemy import text

evil = "x' OR '1'='1"
with engine.connect() as connection:
    unsafe = connection.execute(text(f"SELECT name FROM prompts WHERE name = '{evil}'")).all()
    safe = connection.execute(text("SELECT name FROM prompts WHERE name = :name"), {"name": evil}).all()
    orm = connection.execute(select(Prompt.name).where(Prompt.name == evil)).all()
print("f-string:", len(unsafe), "rows | bound parameter:", len(safe), "rows | ORM:", len(orm), "rows")

f-string: 3 rows | bound parameter: 0 rows | ORM: 0 rows

pg_dump --format=custom --file=prompts.dump "$DATABASE_URL"       # back up (use a plain postgresql:// URL here)
createdb prompts_restore_check
pg_restore --dbname=prompts_restore_check prompts.dump           # restore into a scratch database, then look inside

# 1. Easy
with SessionLocal() as session:
    stmt = (select(Prompt.name, func.count(PromptVersion.id))
            .join(PromptVersion).group_by(Prompt.name).order_by(Prompt.name))
    for name, versions in session.execute(stmt):
        print(name, versions)


# 2. Medium
from app.schemas import PromptIn
from app.service import NotFound, PromptService


def restore(service: PromptService, prompt_id: int, version: int):
    prompt = service.get(prompt_id)
    old = next((v for v in prompt.versions if v.version == version), None)
    if old is None:
        raise NotFound(f"Prompt {prompt_id} has no version {version}")
    return service.new_version(prompt_id, PromptIn(name=prompt.name, template=old.template,
                                                   temperature=old.temperature))


with SessionLocal() as session:
    service = PromptService(session)
    restored = restore(service, prompt_id, 1)
    print([(v.version, v.template) for v in restored.versions])
    try:
        restore(service, prompt_id, 9)
    except NotFound as exc:
        print("NotFound:", exc)

classify-ticket 1
summarise 1
support-answer 3
[(1, 'Answer: {input}'), (2, 'A saves first {input}'), (3, 'Answer politely: {input}'), (4, 'Answer: {input}')]
NotFound: Prompt 3 has no version 9

# file: tests/test_filter.py
def test_percent_is_not_a_wildcard(client):
    client.post("/v1/prompts", json={"name": "support-answer", "template": "x {input}"})
    assert client.get("/v1/prompts?q=%25").json()["items"] == []
    assert len(client.get("/v1/prompts?q=SUPPORT").json()["items"]) == 1

DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/p10_prompts_test uv run python -m pytest -q 2>&1 | tail -3

...                                                                                          [100%]
3 passed in 0.40s