Result Serialisation — Mapping PostgreSQL Types to Python

Moving data between PostgreSQL and Python involves type conversion at every boundary. PostgreSQL’s type system is richer than Python’s — it has UUIDs, TIMESTAMPTZ with timezone information, NUMERIC with arbitrary precision, JSONB, arrays, and custom ENUM types. Each driver (psycopg2, asyncpg) handles these conversions differently, and each has edge cases. Getting these conversions right is the difference between a FastAPI endpoint that returns correct data and one that crashes with a serialisation error or silently truncates precision. This lesson covers the full type mapping and the common conversion pitfalls.

Core Type Mappings #

Handling UUIDs, JSONB and Arrays #

import asyncpg
import json
import uuid

async def setup_connection(conn: asyncpg.Connection):
    """Register codecs for automatic type conversion."""
    # Auto-parse JSONB to Python dict
    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"
    )

# Create pool with init function
pool = await asyncpg.create_pool(
    dsn=DATABASE_URL,
    init=setup_connection   # called for every new connection in pool
)

# Now JSONB columns come back as dicts automatically:
row = await conn.fetchrow("SELECT id, metadata FROM posts WHERE id = $1", 1)
print(type(row["metadata"]))   # dict (not str, thanks to codec)
print(row["metadata"]["reading_time"])   # 5

# UUID handling — asyncpg returns UUID objects
row = await conn.fetchrow("SELECT public_id FROM posts WHERE id = $1", 1)
print(type(row["public_id"]))   # <class 'uuid.UUID'>
print(str(row["public_id"]))    # "a8098c1a-f86e-11da-bd1a-00112444be1e"

# To insert a UUID:
await conn.execute(
    "INSERT INTO posts (public_id, title) VALUES ($1, $2)",
    uuid.uuid4(),   # asyncpg accepts UUID objects directly ✓
    "New Post"
)

Converting Rows to Pydantic Models #

from pydantic import BaseModel, ConfigDict
from datetime import datetime
from uuid import UUID

class PostResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id:         int
    public_id:  UUID
    title:      str
    status:     str
    view_count: int
    created_at: datetime

# From SQLAlchemy ORM object (from_attributes=True needed)
post = session.get(Post, 1)
response = PostResponse.model_validate(post)   # reads attributes via from_attributes

# From asyncpg Record (convert to dict first)
row = await conn.fetchrow("SELECT id, public_id, title, status, view_count, created_at FROM posts WHERE id = $1", 1)
response = PostResponse.model_validate(dict(row))   # dict() converts Record

# From psycopg2 RealDictRow (already dict-like)
cur.execute("SELECT id, public_id, title, status, view_count, created_at FROM posts WHERE id = %s", (1,))
row = cur.fetchone()
response = PostResponse.model_validate(dict(row))

Custom ENUM Type Handling #

from enum import Enum
from pydantic import BaseModel

# Python enum that matches PostgreSQL ENUM
class PostStatus(str, Enum):
    draft     = "draft"
    published = "published"
    archived  = "archived"

class Post(BaseModel):
    id:     int
    title:  str
    status: PostStatus   # Pydantic validates "published" → PostStatus.published

# psycopg2: ENUM comes back as str, Pydantic coerces to PostStatus
# asyncpg: ENUM also comes back as str
row = await conn.fetchrow("SELECT id, title, status FROM posts WHERE id = $1", 1)
post = Post.model_validate(dict(row))
print(post.status)            # PostStatus.published
print(post.status == "published")  # True (str Enum)
print(post.status.value)      # "published"

Common Mistakes #

Mistake 1 — asyncpg JSONB returned as string #

❌ Wrong — treating JSONB as already-parsed dict:

row = await conn.fetchrow("SELECT metadata FROM posts WHERE id = $1", 1)
tags = row["metadata"]["tags"]   # TypeError: string indices must be integers!

✅ Correct — register JSON codec OR parse manually:

tags = json.loads(row["metadata"])["tags"]   # manual parse ✓
# Or register codec as shown above to avoid this everywhere

Mistake 2 — Decimal not JSON serialisable #

❌ Wrong — NUMERIC column causes JSON serialisation error:

row = {"price": Decimal("9.99")}
json.dumps(row)   # TypeError: Object of type Decimal is not JSON serializable

✅ Correct — use FastAPI’s response (handles Decimal) or convert explicitly:

row = {"price": float(Decimal("9.99"))}   # ✓ or use str() to preserve precision

Mistake 3 — Naive datetime from TIMESTAMP (not TIMESTAMPTZ) #

❌ Wrong — TIMESTAMP without timezone gives naive datetime:

created_at TIMESTAMP DEFAULT NOW()   -- no timezone info!

post.created_at.tzinfo   # None — no timezone
post.created_at.isoformat()   # "2025-08-06T14:30:00" — no +00:00

✅ Correct — always use TIMESTAMPTZ:

created_at TIMESTAMPTZ DEFAULT NOW()   -- ✓ UTC + timezone info