Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -446,10 +446,22 @@ Install with the `db` extra: `pip install pgdevkit[db]`.
`SELECT` privileges, and consider `statement_timeout=at_most(...)`.
- **`ComplexHelper`** — handles composite, enum and JSONB columns for the CRUD functions: pass
`complex_helper=ComplexHelper(con)` (optionally with `normalizers`), or `complex_helper="auto"` for a default one,
to any `pg_*` function that takes the argument. What it learns from the catalog (types, column types, adapter
registrations) is cached per connection, so a helper created per call queries the catalog only the first time a
connection is used (the SELECT/INSERT itself is then the only query). The cache cannot see later DDL, nor DDL that
was rolled back after it had been cached: call `helper.clear_cache()` after `CREATE TYPE`/`ALTER TYPE`/column
to any `pg_*` function that takes the argument. **Nothing is registered on the connection:** psycopg's type
registration is global to a connection (and so to the next user of a pooled one: a plain `SELECT` of an enum would
return psycopg's auto-numbered `Enum` member instead of the label, #54). The helpers therefore register the
enum/composite adapters only on the *cursor* that runs a write statement (`pg_insert`, `pg_update`/`pg_update_dict`,
`pg_upsert`/`pg_upsert_dict`, the `*_many` variants), and only for the columns where a dict/list value had to
be converted. Reads (`pg_retrieve*`) select such columns as `to_jsonb(...)` and need no registration. After any
`pg_*` call the connection reads enums as label strings and composites as text (e.g. `'(1,2)'`), as before the call.
`RETURNING` rows of `pg_insert`/`pg_update_dict`/`pg_upsert_dict` (and so `pg_update`/`pg_upsert`) select such
columns as `to_jsonb(...)` too, so with a helper they are plain Python values (a composite as a dict, an enum as its
label) whichever columns were written; without one it is a plain `RETURNING *`.
If you call `recursive_convert` yourself, register the types on the cursor that sends the values, with the `info`
from `load_all_complex_types`/`load_complex_type`: `await helper.register_on(cur, info)`, then execute the converted
value on that same `cur` (passing the connection instead would restore psycopg's global behaviour).
What it learns from the catalog (types and column types) is cached per connection, so a helper created per call
queries the catalog only the first time a connection is used (the SELECT/INSERT itself is then the only query). The cache cannot see later DDL, nor
DDL that was rolled back after it had been cached: call `helper.clear_cache()` after `CREATE TYPE`/`ALTER TYPE`/column
changes or rollbacks of those (typical for tests that create types inside a rolled-back transaction on a reused
connection). `ComplexHelper(con, cache=False)` gives the helper a private cache and the pre-cache behaviour.
- **CRUD functions** — `pg_retrieve`, `pg_retrieve_many` (equality `filters`, plus optional `where=` as a t-string or
Expand Down
113 changes: 90 additions & 23 deletions pgdevkit/db/complex_types.py
Original file line number Diff line number Diff line change
@@ -1,10 +1,14 @@
from __future__ import annotations

import warnings
import weakref
from dataclasses import dataclass, field
from typing import Any, Callable

from psycopg import AsyncConnection
from psycopg import adapters as _psycopg_adapters
from psycopg.abc import AdaptContext
from psycopg.adapt import AdaptersMap
from psycopg.rows import dict_row
from psycopg.sql import Identifier
from psycopg.types.composite import CompositeInfo, register_composite
Expand All @@ -28,12 +32,11 @@ class JsonbArray:
@dataclass
class _TypeCache:
"""What a `ComplexHelper` has learned about one connection's database: the catalog's complex types, the
CompositeInfo/EnumInfo objects built from them (they carry that database's OIDs), which of those are registered
on the connection (an adapter registration lives and dies with the connection), and each table column's type."""
CompositeInfo/EnumInfo objects built from them (they carry that database's OIDs) and each table column's type.
Never "what is registered where": nothing is registered on the connection (see `ComplexHelper.register_on`)."""

system_complex_type_dict: dict[Any, tuple[str, str]] | None = None
complex_types: dict[tuple[str, str], CompositeInfo | EnumInfo] = field(default_factory=dict)
registered: set[CompositeInfo | EnumInfo] = field(default_factory=set)
columns: dict[tuple[tuple[str, str], bool], dict[str, ComplexTypeInfo]] = field(default_factory=dict)
column: dict[tuple[tuple[str, str], str], ComplexTypeInfo] = field(default_factory=dict)
shared: bool = False # whether other helpers on the same connection use it too (then column lookups are cached)
Expand All @@ -52,20 +55,43 @@ def _cache_for(con: AsyncConnection) -> _TypeCache:
return _TypeCache()


def _make_composite_object(values: Any, info: CompositeInfo) -> Any:
# One module-level function, so psycopg's cache of generated loader classes (keyed on it) hits on re-registration
# instead of growing by two classes per registration.
assert info.python_type is not None
return info.python_type(*values)


class _Scope:
"""An `AdaptContext` of its own (a copy of psycopg's defaults, shared with nothing)."""

connection = None

def __init__(self) -> None:
self.adapters = AdaptersMap(_psycopg_adapters)


class ComplexHelper:
"""psycopg adapter for PostgreSQL composite types, enums, and JSONB.

Detects a table's non-scalar columns (composite types, enums, JSONB) and
converts plain dict/list Python values into the psycopg-registered types
those columns need, recursing into nested composite fields.
converts plain dict/list Python values into the Python objects (enum members, composite namedtuples)
psycopg sends for those columns, recursing into nested composite fields.

`normalizers` lets a caller reshape a composite value before conversion,
keyed by composite type name (e.g. a project with a `locale_labels`
composite type that needs locale-key backfilling before it's built) —
this is intentionally the only project-specific extension point; nothing
else about a project's types is hardcoded here.

What it learns from the catalog (the types, the columns' types, which ones are registered on the connection) is
**It never registers anything on the connection.** psycopg's adapter registration is global for the connection
(and so for the next user of a pooled connection: a plain `SELECT` of an enum would start returning psycopg's
`Enum` members instead of the label, see issue #54). What `recursive_convert` returns is sent through a cursor
that has the types registered for itself only (`register_on(cursor, info)`; the `pg_*` helpers do this for the
cursor that runs their statement), and the connection behaves exactly as before. Reads need no registration at
all: the `pg_retrieve*` helpers select complex columns as `to_jsonb(...)`.

What it learns from the catalog (the types and the columns' types) is
cached per connection, for as long as that connection lives: a helper created for every call, on a connection
that is reused (e.g. from a pool), only queries the catalog the first time. The cache can't see schema changes
made afterwards (`CREATE TYPE`, `ALTER TYPE ... ADD VALUE`, altered columns), nor DDL that was rolled back after
Expand Down Expand Up @@ -106,17 +132,29 @@ def complex_types(self, value: dict[tuple[str, str], CompositeInfo | EnumInfo])

@property
def registered(self) -> set[CompositeInfo | EnumInfo]:
return self._cache.registered
"""Deprecated: nothing is registered on the connection any more, so this is always empty (a new set each
time). Use `register_on(cursor, info)` to register types for a statement."""
warnings.warn(
"ComplexHelper.registered is deprecated: types are no longer registered on the connection; "
"use `await helper.register_on(cursor, info)` to register them on a cursor",
DeprecationWarning,
stacklevel=2,
)
return set()

@registered.setter
def registered(self, value: set[CompositeInfo | EnumInfo]) -> None:
self._cache.registered = value
warnings.warn(
"ComplexHelper.registered is deprecated and ignored (types are no longer registered on the connection; "
"see `register_on`)",
DeprecationWarning,
stacklevel=2,
)

def clear_cache(self) -> None:
"""Forget what was learned about this connection's database (see the class docstring)."""
self._cache.system_complex_type_dict = None
self._cache.complex_types.clear()
self._cache.registered.clear()
self._cache.columns.clear()
self._cache.column.clear()

Expand All @@ -125,7 +163,10 @@ async def load_complex_type_dict(self) -> None:
await cur.execute("""
SELECT t.oid,
pg_catalog.format_type ( t.oid, NULL ) AS obj_name,
t.typtype
-- an array type reports its element type's typtype, so an enum[] is known to be an enum
CASE WHEN t.typcategory = 'A'
THEN ( SELECT et.typtype FROM pg_catalog.pg_type et WHERE et.oid = t.typelem )
ELSE t.typtype END AS typtype
FROM pg_catalog.pg_type t
JOIN pg_catalog.pg_namespace n
ON n.oid = t.typnamespace
Expand Down Expand Up @@ -154,9 +195,7 @@ async def _load_complex_type_from_colinfos(self, res: dict[str, Any] | None) ->
return None
udt_schema: str = res["udt_schema"]
udt_name: str = res["udt_name"]
c = await self._get_complex_type(f"{udt_schema}.{udt_name}", res["is_enum"], self.con)
await self._recurse_register(c, self.con)
return c
return await self._get_complex_type(f"{udt_schema}.{udt_name}", res["is_enum"], self.con)

async def load_all_complex_types(
self, table_name: tuple[str, str], include_generated: bool = False
Expand All @@ -179,7 +218,8 @@ async def load_all_complex_types(
from information_schema.columns c
left join enum_types e on e.enum_schema=c.udt_schema
and (e.enum_name=c.udt_name or (c.data_type='ARRAY' and c.udt_name='_'||e.enum_name))
where table_schema=%(schema)s and table_name = %(tbl)s and (is_generated <> 'ALWAYS' or %(include_generated)s)"""
where table_schema=%(schema)s and table_name = %(tbl)s and (is_generated <> 'ALWAYS' or %(include_generated)s)
order by c.ordinal_position"""
async with self.con.cursor(row_factory=dict_row) as cur:
await cur.execute(
colquery,
Expand Down Expand Up @@ -236,24 +276,51 @@ async def _get_complex_type(self, name: str, is_enum: bool, con: AsyncConnection
if (schema, type_name) not in self.complex_types:
ci = await (EnumInfo if is_enum else CompositeInfo).fetch(con, Identifier(schema, type_name))
assert ci is not None, f"{'Enum' if is_enum else 'Complex'} type {name} not found in database"
# `recursive_convert` needs the Python types (`ci.enum`, `ci.python_type`) that psycopg only creates when
# a type is registered: do that in a scratch scope that no connection or cursor ever sees.
scratch = _Scope()
if isinstance(ci, EnumInfo):
register_enum(ci, scratch)
else:
register_composite(ci, scratch, make_object=_make_composite_object)
self.complex_types[(schema, type_name)] = ci
return self.complex_types[(schema, type_name)]

async def _recurse_register(self, info: CompositeInfo | EnumInfo, con: AsyncConnection) -> None:
async def register_on(self, context: AdaptContext, info: ComplexTypeInfo) -> None:
"""Register the psycopg adapters for `info` (and every composite/enum nested in it) in `context`, so that
what `recursive_convert` returns for it can be sent by statements run through that context, and values of
those types come back as those Python objects.

`info` is what `load_all_complex_types(table)` / `load_complex_type(table, column)` return for the column
you are writing, and the value `recursive_convert(value, info, con)` returns must be executed on the same
`context` (the cursor) that you registered it on.

Pass the **cursor** that runs the statement: psycopg gives each cursor its own copy of the adapters, so the
registration lasts as long as the cursor and nothing else on the connection (hence on a pooled one) is
affected. Passing the connection itself registers globally for that connection, for good (psycopg's
default); that is not what the `pg_*` helpers do. A no-op for JSONB (`info` being `Jsonb`/`JsonbArray`) and
for `None`."""
if isinstance(info, (CompositeInfo, EnumInfo)):
if self.system_complex_type_dict is None:
await self.load_complex_type_dict()
await self._recurse_register(info, context, set())

async def _recurse_register(
self, info: CompositeInfo | EnumInfo, context: AdaptContext, seen: set[CompositeInfo | EnumInfo]
) -> None:
assert self.system_complex_type_dict is not None, "System complex type dictionary not loaded"
if info not in self.registered:
if isinstance(info, EnumInfo):
register_enum(info, con)
else:
register_composite(info, con)
self.registered.add(info)
if info in seen:
return
seen.add(info)
if isinstance(info, EnumInfo):
register_enum(info, context, enum=info.enum)
return
register_composite(info, context, factory=info.python_type, make_object=_make_composite_object)
for t in info.field_types:
if t in self.system_complex_type_dict:
name, typtype = self.system_complex_type_dict[t]
ci = await self._get_complex_type(name, typtype == "e", con)
await self._recurse_register(ci, con)
ci = await self._get_complex_type(name, typtype == "e", self.con)
await self._recurse_register(ci, context, seen)

async def recursive_convert(
self,
Expand Down
Loading
Loading