Asyncpg Live Vs Mock Shape

asyncpg-live-vs-mock-shape

✅ PROMOTED: Rename + expansion of the previously-promoted asyncpg-decimal-test-shape skill. TDD pressure-test Cycle 2 (JSONB extension) PASS. The RED-Subagent used a dict mock and flagged the codec risk heuristically. The GREEN-Subagent (with skill) used a json.dumps(...) str-mock correctly and delivered 2-layer defense (defensive json.loads + set_type_codec). This skill prevents the "green-test-red-production" class of bugs for Decimal aggregates (Class A) AND JSONB/UUID/INET types (Classes B-E). Auto-discoverable.

Overview

asyncpg returns PostgreSQL types as Python types. The mappings are often different from what naive mock intuition expects:

Mock test green, live production code red — the class of bugs this skill prevents.

Class A — Numeric Aggregates (Decimal)

PostgreSQL aggregates (SUM, COUNT, AVG, STDDEV) on numeric or bigint columns come back via asyncpg as decimal.Decimal, NOT as int or float.

Typical symptom:

TypeError: unsupported operand type(s) for /: 'float' and 'decimal.Decimal'

Quick lookup PG aggregate → asyncpg Python type

Mock pattern

Wrong (passes shape-test, fails live):

return {
    "day_pnl_eur": 50.0,    # float — wrong, real is Decimal
    "r30_count":  5,         # int   — wrong, real is Decimal
    "r30_wins":   3,         # int   — wrong, real is Decimal
}

Correct:

from decimal import Decimal
return {
    "day_pnl_eur": Decimal("50.0"),
    "r30_count":  Decimal(5),
    "r30_wins":   Decimal(3),
}

Defensive code layer (production)

r30_count = float(r["r30_count"] or 0)
r30_wins  = float(r["r30_wins"]  or 0)
wr30 = (r30_wins / r30_count) * 100 if r30_count > 0 else None

The skill recommends both: Decimal in the mock AND float-cast in the code.

Class B — JSONB / JSON (str, NOT dict)

SELECT some_jsonb_column FROM ... with asyncpg returns the raw JSON text as str, not as a parsed Python dict. Symptoms vary by access pattern:

SQL example

-- ml_models has report JSONB; query returns raw JSON string
SELECT id, report FROM ml_models WHERE active = true;

row = await conn.fetchrow("SELECT report FROM ml_models WHERE id = $1", model_id)
report = row["report"]
# ⚠️ report is str, not dict!
# report["confidence"]   → TypeError: string indices must be integers

Mock pattern for JSONB

Wrong (passes shape-test, fails live):

return {
    "id": 713,
    "report": {"confidence": 0.75, "verdict": "accept"},   # ← dict, real is str!
}

Correct:

import json
return {
    "id": 713,
    "report": json.dumps({"confidence": 0.75, "verdict": "accept"}),  # ← str
}

Defensive code layer (production)

Two options — both good, depending on codebase convention:

Option 1 — Defensive json.loads per call:

raw = row["report"]
report = json.loads(raw) if isinstance(raw, str) else raw  # handles already-dict (e.g. test mock that uses dict)
confidence = report["confidence"]

Option 2 — set_type_codec at the pool-init level (clean structural solution):

async def init_pool():
    pool = await asyncpg.create_pool(...)
    async with pool.acquire() as conn:
        await conn.set_type_codec(
            "jsonb",
            encoder=json.dumps,
            decoder=json.loads,
            schema="pg_catalog",
        )
        await conn.set_type_codec(
            "json",
            encoder=json.dumps,
            decoder=json.loads,
            schema="pg_catalog",
        )
    return pool

⚠️ With set_type_codec: the mock then also returns dict and that is consistent. BUT: check ALL places where the pool is used (some tests may use a separate pool without the codec → divergent mock shape depending on test setup).

When it occurs

• Live connection via asyncpg.connect() or asyncpg.create_pool() without JSONB codec → str
• With set_type_codec(jsonb, decoder=json.loads) → dict
• psycopg2 with json.loads cursor → dict (different behavior)

Example

An ML-evaluator endpoint read ml_models.report (JSONB) and tried report["confidence"] directly — TypeError: string indices must be integers. The mock test was green with a dict mock. Fix in two steps:

1. set_type_codec at pool-init (structurally clean)
2. Defensive json.loads in the specific function (belt-and-suspenders)

Class C — UUID

SELECT id FROM x with an id uuid column returns a uuid.UUID object, not str. Symptoms:

# mock returns str: "abc-123-..."
# live returns UUID("abc-123-...")
row["id"] == "abc-123-..."   # ← False on live (UUID != str comparison)

Mock pattern for UUID

Wrong:

return {"id": "550e8400-e29b-41d4-a716-446655440000", ...}

Correct:

import uuid
return {"id": uuid.UUID("550e8400-e29b-41d4-a716-446655440000"), ...}

Defensive code

str(row["id"])   # works for UUID and for str
# or
row["id"] if isinstance(row["id"], str) else str(row["id"])

Class D — INET / CIDR

SELECT ip FROM x with an ip inet column returns an ipaddress.IPv4Address / IPv6Address / IPv4Network / IPv6Network object, not str. Symptom: JSON serialization fails (object not serializable).

Mock pattern for INET

Wrong:

return {"client_ip": "192.168.1.1"}

Correct:

import ipaddress
return {"client_ip": ipaddress.IPv4Address("192.168.1.1")}

Defensive code

str(row["client_ip"])  # works for IPv4Address and for str

Class E — Timestamps (Aware vs Naive)

timestamp without time zone → naive datetime. timestamptz → aware datetime with tzinfo. Comparing aware vs naive raises TypeError.

Mock pattern

When the DB column is timestamptz:

from datetime import datetime, timezone
return {"created_at": datetime(2026, 5, 29, 8, 0, tzinfo=timezone.utc)}

When the DB column is timestamp (without TZ):

return {"created_at": datetime(2026, 5, 29, 8, 0)}   # NAIVE

Production code SHOULD typically keep all timestamps aware — if the DB column is timestamp and the code does aware comparison → a DB schema fix is cleaner than a defensive workaround.

When to apply — trigger table

Anti-Patterns

• ❌ Test with int/float mocks → green pytest, red live (Class A)
• ❌ Test with dict mock for JSONB column → green pytest, string indices must be integers live (Class B)
• ❌ Test with str mock for UUID column → green pytest, comparison fail live (Class C)
• ❌ Assuming "asyncpg already coerces to whatever fits" — no, raw driver, no ORM
• ❌ set_type_codec inconsistent across pools (some test pools have it, others don't) → different mock shape depending on pool-init

Defense-in-depth recommendation

Cross-references

• The "code review must become standard" maxim — reviewers did not find the Decimal-mock bug + JSONB-mock bug locally, because the mocks were green
• superpowers:test-driven-development — base pattern that this skill is only a test-shape refinement of

Background: TDD log (bulletproofing log)

Cycle 1 (PASS) — Decimal-only skill

• RED-Subagent (without skill): used int for COUNT, float for SUM/AVG. Flagged uncertainty explicitly. Test would be green, live would be TypeError.
• GREEN-Subagent (with skill): used Decimal for all aggregates + type assertions + defensive-coercion note. Skill self-reflection called the PG→asyncpg lookup table a "killer feature".
• Verdict: PROMOTE Decimal-only skill. Skill delivers demonstrable bug avoidance.

Cycle 2 (PASS — JSONB extension)

Scenario: A FastAPI endpoint loads ml_models.report (JSONB) and does row["report"]["confidence"] directly. Pytest-mock task — RED without skill / GREEN with skill.

• RED-Subagent (without skill): used dict for the report mock ("JSONB is decoded as dict by asyncpg by default"). Self-reflection was unusually honest — explicitly flagged: "If the pool has not registered a JSON codec, report comes back as str, and row["report"]["confidence"] throws TypeError. That is a real live-bug risk that my mock test masks." RED recognized the bug HEURISTICALLY, but would not have corrected the mock without pool-setup verification.
• GREEN-Subagent (with skill): used json.dumps(report_payload) directly, Class-B table as a quick lookup, both defense-in-depth options documented (defensive json.loads + set_type_codec), production-code fix provided as a comment. Skill self-reflection called the Class-B table "killer content". GREEN code was safe + traceable where RED code was "plausible but with latent risk".
• Verdict: PROMOTE. Skill prevents exactly the JSONB bug class. RED's self-honesty showed that even a careful engineer recognizes-but-does-not-fix the bug without an external reference.

Refactor applied (inline before promote):

• Class-B symptom table: 5 access patterns → exact live errors (previously a terse 3-line list). GREEN self-reflection feedback on symptom clarity incorporated.

Cycle-2 backlog (polish, non-blocking after PROMOTE)

1. conn.fetch() (list of rows) example in addition to fetchrow — more frequent with window functions
2. AsyncMock + pool.acquire() context-manager plumbing: separate skill candidate asyncpg-pool-mock-plumbing (GREEN suggestion) — orthogonal to the type-shape question, a recurring stumbling block
3. Empty-set behavior column for the Class-A lookup table: COUNT(*) → Decimal(0), SUM/AVG → None for empty result set
4. set_type_codec pattern library: document the repo-wide convention — when codec (all pools consistent), when defensive
5. Class F (arrays): text[]/int[] → list[str]/list[int] (usually ok but check the dimension assumption)
6. Class G (empty JSONB / NULL): report IS NULL → asyncpg returns None, defensive or {} pattern
7. pytest.approx for Decimal comparisons note — not needed if production code finally casts to float

Real-world impact

Class A: An equity-curve endpoint mock test with int/float → 4/4 green; live deploy → 500 Internal Server Error in 5s. Skill would catch it locally.

Class B: An ML-evaluator endpoint mock test with dict → 32 tests green; live run → string indices must be integers. Fix in 2 steps: set_type_codec + defensive json.loads. Both hits in one session triggered this skill extension.

Skill would catch both bug classes locally without a live roundtrip.