Phoenix Server

Phoenix Backend Development

Phoenix is an AI observability platform. The backend is Python: FastAPI serving a REST API and Strawberry GraphQL API over an async SQLAlchemy ORM (PostgreSQL + SQLite).

Development Guide Index

Read DEVELOPMENT.md (env setup, uv, tests, debugpy, pre-commit, REST API conventions) and CONTRIBUTING.md (PR format, conventional commits, code review expectations) if you have not already.

Everyday Commands

make dev-backend                        # backend only, no frontend build needed
uv run pytest path/to/test -n auto      # run specific tests in parallel
make test-python                        # full test suite
make graphql                            # regenerate schema after GQL changes
make format                             # format all code
make typecheck-python                   # mypy + pyright

Key Directories

src/phoenix/server/api/
  mutations/        Domain-specific mutation mixins, composed in __init__.py
  types/            GraphQL types with field resolvers
  input_types/      Strawberry @input classes with validation
  subscriptions.py  Async generator subscriptions (streaming)
  queries.py        Root query type
  context.py        Request context: db, dataloaders, auth, event queue
  dataloaders/      Batch loaders (prevent N+1 queries)
  auth.py           Permission classes (IsNotReadOnly, IsNotViewer, etc.)
  routers/          REST API endpoints (v1/)
src/phoenix/db/
  models.py         SQLAlchemy ORM models (single file)
  migrations/       Alembic migrations
tests/unit/server/api/
  mutations/        Mutation tests
  types/            Type resolver tests
  conftest.py       Fixtures: db, gql_client, test data factories

What Are You Doing?

Hard Rules

• Side effects belong on `Mutation`, not `Query`. A resolver that makes outbound

network calls, reads secrets, writes state, or accepts a user-supplied URL/host MUST be a @strawberry.mutation with permission_classes=[...]. Query fields bypass the make check-graphql-permissions CI guard and are reachable unauthenticated by default — this has been exploited as an SSRF vector. See references/graphql-patterns.md → "Query vs Mutation".

Naming

• Avoid acronyms and single/double-letter abbreviations for local variables.

Prefer the full noun: session / project_session over ps, trace over t, example / dataset_example over de. The cost of a longer identifier is trivial; the cost of having to mentally expand an acronym while reading unfamiliar code is not.

• Established domain acronyms used in the codebase (db, gql, otel, llm)

are fine — they're vocabulary, not abbreviations of local nouns.

Docstrings

The project rule of "default to no comments" is about inline comments, not docstrings. Public APIs should be documented.

• Document parameters and return values on public methods of reusable classes

(clients, services, factories, builders). Use Google-style Args: / Returns: / Raises: blocks when the meaning isn't fully recoverable from the type signature. Do not strip these during refactors — semantics outlive file moves.

• Describe behavior, not implementation. A method on a docs-search client

says "Invoke a backend tool and return its text result", not "Invoke a tool on the MCP server" — the underlying transport is an implementation detail and the docstring should survive a transport swap. Internal helpers (leading _) may reference the transport directly since their scope is bounded.

• One-liner docstrings are fine when the name and types fully convey intent

(close(), is_backend_tool(name)). Don't pad them with restated signatures.

• Module docstrings belong at the top of any file that exposes public

surface (a client class, a router, a service module). One sentence on what the module is for is enough.

Database Patterns

Patterns for database models, Alembic migrations, and async session management.

Adding a Migration

Always scaffold migrations with Alembic — never write migration files from scratch:

uv run alembic revision -m "add things table"

This creates a new file in src/phoenix/db/migrations/versions/ with the revision chain already wired up. Then fill in upgrade() and downgrade().

batch_alter_table (Required for SQLite Compatibility)

SQLite has limited ALTER TABLE support. All column additions, drops, and constraint changes must use batch_alter_table:

def upgrade() -> None:
    with op.batch_alter_table("things") as batch_op:
        batch_op.add_column(sa.Column("status", sa.String, nullable=True))

def downgrade() -> None:
    with op.batch_alter_table("things") as batch_op:
        batch_op.drop_column("status")

For multi-step operations (add column, backfill, add constraint):

def upgrade() -> None:
    # Step 1: Add column as nullable
    with op.batch_alter_table("things") as batch_op:
        batch_op.add_column(sa.Column("status", sa.String, nullable=True))

    # Step 2: Backfill
    op.execute("UPDATE things SET status = 'active' WHERE status IS NULL")

    # Step 3: Add constraints
    with op.batch_alter_table("things") as batch_op:
        batch_op.alter_column("status", nullable=False, existing_nullable=True)
        batch_op.create_check_constraint(
            "valid_status",
            "status IN ('active', 'inactive')",
        )

JSONB Shim (Must Be Redefined Per Migration)

Migrations must be self-contained — they can't depend on the current state of models.py or any other source file which changes over time. When custom ORM types are needed inside a migration (e.g., for JSONB types), do not import those types from other modules in the codebase, but instead copy and re-implement inside the migration file itself to ensure it is self-contained.

Create Table Example

def upgrade() -> None:
    op.create_table(
        "things",
        sa.Column("id", sa.Integer, primary_key=True),
        sa.Column("name", sa.String, nullable=False, unique=True),
        sa.Column("description", sa.Text, nullable=True),
        sa.Column("metadata_", JSON_, nullable=False, server_default="{}"),
        sa.Column(
            "created_at",
            sa.TIMESTAMP(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
        ),
        sa.Column(
            "updated_at",
            sa.TIMESTAMP(timezone=True),
            nullable=False,
            server_default=sa.func.now(),
            onupdate=sa.func.now(),
        ),
    )

Model Patterns

Models live in src/phoenix/db/models.py (single file).

Basic Model

class Thing(Base):
    __tablename__ = "things"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, nullable=False, unique=True)
    description: Mapped[Optional[str]] = mapped_column(Text, nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        UtcTimeStamp, server_default=func.now()
    )
    updated_at: Mapped[datetime] = mapped_column(
        UtcTimeStamp, server_default=func.now(), onupdate=func.now()
    )

Session Management

In Mutations/Resolvers

async with info.context.db() as session:
    thing = models.Thing(name="foo")
    session.add(thing)
    # Auto-commits on clean exit of the async with block

In Tests

The db fixture (type DbSessionFactory) is provided by conftest.py and gives each test an isolated transaction:

async with db() as session:
    thing = await session.get(models.Thing, thing_id)
    assert thing is not None

Key Rules

• One session per `async with` block — don't nest or share sessions across blocks.
• `await session.flush()` when you need auto-generated IDs before the block exits.
• The session expires after the block — don't access ORM attributes on objects after

the async with exits unless they were eagerly loaded or detached.

GraphQL Patterns

Full templates and patterns for mutations, types, subscriptions, and input types.

Query vs Mutation: Side Effects MUST Be Mutations

A GraphQL field belongs on Mutation (not Query) if it does any of the following:

• Makes outbound HTTP, gRPC, or other network calls to user-controlled or

user-influenced destinations (URLs, hostnames, headers, kwargs)

• Calls a "test connection" / "validate credentials" / "ping" style operation
• Writes to the database, the filesystem, or an external store
• Reads or decrypts secrets
• Triggers a side effect that costs money, time, or external rate limits

Why this matters: Query fields are conventionally treated as safe, cacheable reads. They are easy to forget on the auth path, they are reachable through introspection, and permission_classes are far less commonly applied to query resolvers than to mutations. A query resolver that issues HTTP requests with a user-supplied base_url is an unauthenticated SSRF vector — this has happened in this codebase (testGenerativeModelCustomProviderCredentials was originally a query and was used to reach internal cloud metadata endpoints).

If you find yourself writing a @strawberry.field on Query that takes a URL, a config object, credentials, or anything that ends up making a network call: stop and make it a mutation with permission_classes=[...]. The CI guard make check-graphql-permissions enforces that all mutations and subscriptions declare permission classes; queries get no such guarantee.

When in doubt, ask: "could an unauthenticated attacker who can reach /graphql abuse this field to do something they should not?" If the answer is anything other than a confident no, it must be a mutation behind auth.

Adding a Mutation

Step-by-step

1. Define an input type (or reuse an existing one) 2. Define a payload type 3. Write the mutation method in a mixin class 4. Register the mixin in mutations/__init__.py 5. Run make graphql to regenerate the schema

Input Type Template

from typing import Optional

import strawberry
from strawberry import UNSET

from phoenix.server.api.exceptions import BadRequest


@strawberry.input
class CreateThingInput:
    name: str
    description: Optional[str] = UNSET  # "not provided" vs "null"

    def __post_init__(self) -> None:
        if not self.name.strip():
            raise BadRequest("Name cannot be empty")

For patch/update inputs, include the target ID:

@strawberry.input
class PatchThingInput:
    id: GlobalID
    name: Optional[str] = UNSET
    description: Optional[str] = UNSET

For union/oneOf inputs (pick exactly one variant):

@strawberry.input(one_of=True)
class ThingConfigInput:
    variant_a: Optional[VariantAInput] = strawberry.UNSET
    variant_b: Optional[VariantBInput] = strawberry.UNSET

Payload Type Template

Some payloads may include query: Query so the client can chain follow-up reads:

from phoenix.server.api.queries import Query


@strawberry.type
class ThingMutationPayload:
    thing: Thing
    query: Query

For delete mutations, return Query. If the deleted entity is a GraphQL node, also return the deleted node ID. This enables the frontend to the @deleteEdge directive to update the Relay cache without refetching dataset from the server.

@strawberry.type
class DeleteThingMutationPayload:
    id: GlobalID
    query: Query

async def delete_thing(self, ...) -> DeleteThingMutationPayload:
    # ... delete ...
    return DeleteThingMutationPayload(
        id=GlobalID("Thing", str(thing_id)),
        query=Query(),
    )

Mutation Mixin Template

import strawberry
from strawberry.types import Info

from phoenix.db import models
from phoenix.server.api.auth import IsLocked, IsNotReadOnly, IsNotViewer
from phoenix.server.api.context import Context
from phoenix.server.api.queries import Query
from phoenix.server.dml_event import ThingInsertEvent


@strawberry.type
class ThingMutationMixin:
    @strawberry.mutation(permission_classes=[IsNotReadOnly, IsNotViewer, IsLocked])
    async def create_thing(
        self,
        info: Info[Context, None],
        input: CreateThingInput,
    ) -> ThingMutationPayload:
        name = input.name.strip()
        description = input.description if input.description is not UNSET else None

        async with info.context.db() as session:
            thing = models.Thing(name=name, description=description)
            session.add(thing)

        # Emit DML event after the session commits
        info.context.event_queue.put(ThingInsertEvent((thing.id,)))

        return ThingMutationPayload(
            thing=to_gql_thing(thing),
            query=Query(),
        )

Registering the Mixin

Add to src/phoenix/server/api/mutations/__init__.py:

from phoenix.server.api.mutations.thing_mutations import ThingMutationMixin

@strawberry.type
class Mutation(
    ThingMutationMixin,
    # ... existing mixins ...
):
    pass

Adding a GraphQL Type

Type Template

from typing import Optional

import strawberry
from strawberry.relay import Node, NodeID
from strawberry.types import Info

from phoenix.db import models
from phoenix.server.api.context import Context


@strawberry.type
class Thing(Node):
    id: NodeID[int]
    db_record: strawberry.Private[Optional[models.Thing]] = None

    def __post_init__(self) -> None:
        if self.db_record and self.id != self.db_record.id:
            raise ValueError("Thing ID mismatch")

    @strawberry.field
    async def name(self, info: Info[Context, None]) -> str:
        if self.db_record:
            return self.db_record.name
        return await info.context.data_loaders.thing_fields.load(
            (self.id, models.Thing.name),
        )

    @strawberry.field
    async def description(self, info: Info[Context, None]) -> Optional[str]:
        if self.db_record:
            return self.db_record.description
        return await info.context.data_loaders.thing_fields.load(
            (self.id, models.Thing.description),
        )

Converter Function

Place near the type definition or in a helpers module:

def to_gql_thing(thing: models.Thing) -> Thing:
    return Thing(id=thing.id, db_record=thing)

Lazy Imports for Circular References

When type A references type B and vice versa, use strawberry.lazy():

from typing import TYPE_CHECKING, Annotated

import strawberry

if TYPE_CHECKING:
    from .OtherType import OtherType


@strawberry.field
async def related(self, info: Info[Context, None]) -> Annotated["OtherType", strawberry.lazy(".OtherType")]:
    from .OtherType import OtherType
    return OtherType(id=related_id)

Polymorphic Types (Interface + Implementations)

When using @strawberry.interface, register all implementing types explicitly in the schema. Check schema.py for the _implementing_types() helper that collects them.

Subscription Pattern

Subscriptions are async generators with concurrency control:

@strawberry.type
class Subscription:
    @strawberry.subscription(permission_classes=[IsNotReadOnly, IsNotViewer, IsLocked])
    async def watch_thing(
        self,
        info: Info[Context, None],
        input: WatchThingInput,
    ) -> AsyncIterator[ThingPayload]:
        try:
            # Setup phase
            async with info.context.db() as session:
                # ... validate, fetch initial state ...
                pass

            # Streaming phase
            while has_more:
                yield ThingPayload(...)
        finally:
            # Cleanup: cancel tasks, close generators, release resources
            # Order matters — cancel first, then await, then close
            pass

Subscription cleanup order is critical: cancel tasks, await cancellation, close generators. Getting this wrong causes resource leaks.

Gotchas

UNSET vs None — UNSET means "the client didn't send this field." None means "the client explicitly sent null." This distinction matters for patch mutations where you need to differentiate "leave unchanged" from "clear the value."

Lazy imports are required for circular type references — Strawberry resolves types eagerly. If type A has a field returning type B and type B has a field returning type A, both must use strawberry.lazy() and TYPE_CHECKING guards.

LLM Trace Tests

Tests for code that emits OpenInference spans against an LLM provider have two distinct concerns:

1. Replay — pin the LLM HTTP exchange so the test is deterministic and offline. 2. Assertion — verify that the right OpenInference span attributes get emitted.

Both have non-obvious gotchas. Follow the patterns below.

VCR cassette workflow

Use `tests/unit/vcr.py` CustomVCR (record_mode "once").

with custom_vcr.use_cassette():
    response = await wrapped_model.request(...)

Cassette path is derived from the test module name + test function name. Renaming either orphans the cassette. Move both together, or delete and re-record.

Recording vs. replay

• First run with no cassette → records by hitting the real provider. Requires a real API key in env.
• Subsequent runs → replay only. SDK still needs some API key to construct its client even though VCR intercepts the HTTP call.

Always wire in the API-key fixture so replay works without real credentials. tests/unit/conftest.py already exposes openai_api_key and anthropic_api_key fixtures that monkeypatch the env var to a fake value:

@pytest.fixture
def wrapped_model(
    tracer_provider: TracerProvider,
    anthropic_api_key: str,
) -> OpenInferenceModelWrapper:
    return OpenInferenceModelWrapper(
        AnthropicModel(MODEL_NAME, provider=AnthropicProvider()),
        tracer_provider=tracer_provider,
    )

Determinism

Free-form LLM output is impossible to assert exactly. Either:

1. Pin via cassette + don't assert on output text (only structure/role/etc.). Brittle if you re-record. 2. Prefer: prompt the model to repeat an exact phrase, set temperature=0.0, and assert exact equality. Survives re-recording.

expected_output = "The capital of France is Paris."
messages = [ModelRequest(parts=[
    SystemPromptPart(content=f"Reply with exactly the following sentence and nothing else: {expected_output}"),
    UserPromptPart(content="What is the capital of France?"),
])]
settings = ModelSettings(temperature=0.0, max_tokens=32)
...
assert response_text == expected_output

Re-recording when the request changes

When you change a prompt or any other field of the outgoing request, the cassette's recorded request body no longer matches and VCR will refuse the call. Always delete the cassette and re-record against the real provider — never hand-edit the YAML.

Hand-editing looks tempting (especially for streaming cassettes where you might want to split text across multiple content_block_delta events), but:

• It silently lets the cassette diverge from any real response shape the provider would actually return, which defeats the point of recording.
• Future schema changes from the provider will surprise you in production but pass in test.
• It's easy to introduce subtle inconsistencies (token counts, IDs, finish reasons) that mask bugs.

If a "Repeat exactly" prompt pinned the output text, re-recording is reproducible. That's the determinism mechanism — not editing the cassette.

Span attribute assertions

Use the pop + assert not attributes pattern

attributes = dict(span.attributes or {})
assert attributes.pop(OPENINFERENCE_SPAN_KIND) == LLM
assert attributes.pop(LLM_PROVIDER) == PROVIDER_ANTHROPIC
# ... pop every attribute you expect ...
assert not attributes

This catches unexpected attributes too. Spot-checking a handful of keys lets bugs slip in (e.g. wrapper accidentally emits a sensitive field).

Use OpenInference semantic-convention constants, not literal strings

from openinference.semconv.trace import (
    MessageAttributes,
    SpanAttributes,
    ToolAttributes,
    ToolCallAttributes,
)

LLM_INPUT_MESSAGES = SpanAttributes.LLM_INPUT_MESSAGES
MESSAGE_ROLE = MessageAttributes.MESSAGE_ROLE
MESSAGE_CONTENT = MessageAttributes.MESSAGE_CONTENT
MESSAGE_TOOL_CALLS = MessageAttributes.MESSAGE_TOOL_CALLS
MESSAGE_TOOL_CALL_ID = MessageAttributes.MESSAGE_TOOL_CALL_ID
TOOL_JSON_SCHEMA = ToolAttributes.TOOL_JSON_SCHEMA
TOOL_CALL_ID = ToolCallAttributes.TOOL_CALL_ID
TOOL_CALL_FUNCTION_NAME = ToolCallAttributes.TOOL_CALL_FUNCTION_NAME
TOOL_CALL_FUNCTION_ARGUMENTS_JSON = ToolCallAttributes.TOOL_CALL_FUNCTION_ARGUMENTS_JSON

assert attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_ROLE}") == "user"
assert attributes.pop(f"{LLM_INPUT_MESSAGES}.0.{MESSAGE_CONTENT}") == prompt

Literal strings like "message.role" work but break silently if the spec moves. Constants are typo-safe.

assert isinstance(...) over cast(...)

OTel attribute values are typed as AttributeValue (a union including str, int, sequences). cast(str, ...) lies to the type checker; isinstance is a real runtime check.

inv_params = attributes.pop(LLM_INVOCATION_PARAMETERS)
assert isinstance(inv_params, str)
assert json.loads(inv_params) == dict(settings)

Works with the walrus operator when you just need the type-narrow + value:

assert isinstance(prompt_tokens := attributes.pop(LLM_TOKEN_COUNT_PROMPT), int)
assert isinstance(completion_tokens := attributes.pop(LLM_TOKEN_COUNT_COMPLETION), int)
assert attributes.pop(LLM_TOKEN_COUNT_TOTAL) == prompt_tokens + completion_tokens

Validate input.value / output.value JSON exhaustively

Don't just do assert "messages" in parsed_input. Assert the full key set and every field that's deterministic:

parsed_input = json.loads(input_value)
assert set(parsed_input) == {"messages", "model_settings", "model_request_parameters"}
assert len(parsed_input["messages"]) == 1
assert parsed_input["messages"][0]["kind"] == "request"
assert parsed_input["model_settings"] == dict(settings)

set(parsed_input) == {...} catches extra fields too — same idea as assert not attributes.

Streaming: assert event accumulation matches final text

If you yield partial text via streaming events, accumulate it during the loop and check it equals the final assembled text:

event_text_chunks: list[str] = []
async for event in stream:
    if isinstance(event, PartStartEvent) and isinstance(event.part, TextPart):
        event_text_chunks.append(event.part.content)
    elif isinstance(event, PartDeltaEvent) and isinstance(event.delta, TextPartDelta):
        event_text_chunks.append(event.delta.content_delta)
final_response = stream.get()
streamed_text = "".join(p.content for p in final_response.parts if isinstance(p, TextPart))
assert "".join(event_text_chunks) == streamed_text == expected_output

Hand-craft the cassette to deliver text across multiple delta events; otherwise the multi-delta accumulation logic isn't exercised.

Gotchas

• Conditional attributes. If the wrapper only sets cache_read/cache_write token attrs when the values are non-zero, and the cassette has zero, those keys won't appear. Don't add a defensive attributes.pop(key, None) "in case" — it hides real regressions. Either drive the keys to non-zero in the cassette and assert them, or accept they won't be present.
• Where the SDK reads its API key. Constructing AnthropicModel(provider=AnthropicProvider()) reads ANTHROPIC_API_KEY at construction time. Wire the API-key fixture into the fixture that builds the model, not just into individual tests.
• `record_mode="once"` doesn't auto-update. Once a cassette exists, VCR will fail with a "request not found" error if the request changes — it won't silently re-record. Delete the cassette to re-record.
• Constants block placement. If many tests share a long block of SpanAttributes.X = ... aliases, push it to the bottom of the test file. Names referenced inside test/fixture bodies resolve at call time, so test ordering doesn't matter.
• Don't construct test helpers around pydantic_ai types. ModelSettings(...) and ModelRequestParameters(...) are direct constructors — no _settings(...) / _empty_request_parameters(...) wrapper helpers needed. Inline the construction in the test.

Reference

A worked example covering all of the above: `tests/unit/server/agents/pydantic_ai/test_openinference_model_wrapper.py`.

Test Patterns

Patterns for writing backend tests against the Phoenix GraphQL API.

Running Tests

make test-python                              # full suite
uv run pytest path/to/test_file.py -n auto    # specific file, parallel
uv run pytest path/to/test_file.py -n auto -x # stop on first failure
uv run pytest path/to/test_file.py -xvs       # verbose, no capture (for debugging)
uv run pytest path/to/test_file.py --run-postgres  # also run against PostgreSQL

-n auto uses pytest-xdist to parallelize across CPU cores. Always use it unless you're debugging a specific test and need sequential output.

Key Fixtures

These come from tests/unit/conftest.py and tests/unit/server/api/conftest.py:

The db fixture provides per-test transaction isolation. SQLite uses in-memory databases with savepoint rollback; PostgreSQL uses template database cloning.

Mutation Test Template

import pytest
from strawberry.relay import GlobalID

from phoenix.db import models
from phoenix.server.types import DbSessionFactory
from tests.unit.graphql import AsyncGraphQLClient


class TestCreateThing:
    _MUTATION = """
      mutation CreateThing($input: CreateThingInput!) {
        createThing(input: $input) {
          thing {
            id
            name
            description
          }
          query { __typename }
        }
      }
    """

    async def test_creates_thing(
        self,
        gql_client: AsyncGraphQLClient,
        db: DbSessionFactory,
    ) -> None:
        # Execute mutation
        result = await gql_client.execute(
            self._MUTATION,
            variables={"input": {"name": "my-thing", "description": "desc"}},
        )

        # Assert GraphQL response
        assert result.data and not result.errors
        thing = result.data["createThing"]["thing"]
        assert thing["name"] == "my-thing"
        assert thing["description"] == "desc"

        # Verify database state
        thing_id = int(GlobalID.from_id(thing["id"]).node_id)
        async with db() as session:
            db_thing = await session.get(models.Thing, thing_id)
            assert db_thing is not None
            assert db_thing.name == "my-thing"

Pattern Notes

Always assert both `result.data` and `not result.errors` — a response can have data and errors simultaneously (partial success). Checking both catches subtle issues.

Use a private `_MUTATION` class constant for the GraphQL string. If a test class has multiple tests against the same operation, they all share the string.

Query Test Template

async def test_lists_things(
    gql_client: AsyncGraphQLClient,
    some_fixture_that_creates_data: None,
) -> None:
    query = """
      query {
        things { edges { node { id name } } }
      }
    """
    result = await gql_client.execute(query=query)
    assert not result.errors
    assert len(result.data["things"]["edges"]) == 3

GlobalID Construction and Parsing

from strawberry.relay import GlobalID

# Construct: type name + string ID
gid = str(GlobalID("Dataset", str(dataset.id)))
# Result: base64-encoded relay ID like "RGF0YXNldDox"

# Parse: extract numeric ID from relay ID
node_id = int(GlobalID.from_id(relay_id_string).node_id)

Data Setup Fixtures

Two patterns for setting up test data:

Pattern 1: Direct ORM Inserts (most common)

@pytest.fixture
async def thing_with_children(db: DbSessionFactory) -> models.Thing:
    async with db() as session:
        thing = models.Thing(name="parent")
        session.add(thing)
        await session.flush()  # flush to get the auto-generated ID

        child = models.Child(thing_id=thing.id, value="child-1")
        session.add(child)
        await session.flush()

    return thing

Use await session.flush() after session.add() when you need the auto-generated ID for subsequent inserts. The session auto-commits when the async with block exits cleanly.

Pattern 2: Complex fixture chains

Fixtures can depend on other fixtures for layered setup:

empty_dataset                          # base dataset + versions
  -> dataset_with_experiments          # adds experiments
    -> dataset_with_experiments_and_runs   # adds run data

Look at tests/unit/server/api/conftest.py for the full fixture chain — it has ready-made fixtures for datasets, experiments, evaluators, and more.

Subscription Test Pattern

async def test_subscription(
    self,
    gql_client: AsyncGraphQLClient,
) -> None:
    async with gql_client.subscription(
        query=self._SUBSCRIPTION,
        variables={"input": {...}},
    ) as sub:
        async for data in sub.stream():
            typename = data["watchThing"]["__typename"]
            if typename == "TextChunk":
                assert data["watchThing"]["content"]
            elif typename == "SubscriptionResult":
                break

Gotchas

Fixture scope is per-test — Each test gets a fresh database transaction that rolls back after the test completes. Don't rely on data from a previous test.