diff --git a/pgdevkit/connection.py b/pgdevkit/connection.py
index 76574d6..1036f06 100644
--- a/pgdevkit/connection.py
+++ b/pgdevkit/connection.py
@@ -7,23 +7,63 @@
     ".database.azuredatabricks.net",
     ".database.cloud.databricks.com",
 )
+_AZURE_POSTGRES_HOST_SUFFIXES = (
+    ".postgres.database.azure.com",
+    ".postgres.cosmos.azure.com",
+)
+
+_default_credential = None
+_managed_identity_credential = None
 
 
 def detect_provider(host: str) -> Literal["azure_postgres", "databricks_lakebase"]:
-    """Classify a Postgres hostname: Databricks Lakebase (needs credential
-    exchange) or the default Azure Postgres AAD token flow."""
+    """Classify a Postgres hostname for Entra ID auth: Databricks Lakebase
+    (needs credential exchange) or the default Azure Postgres AAD token
+    flow — the latter is the fallback for any non-Lakebase host, since
+    `entra_user` is itself the caller's assertion that Entra auth applies.
+    For a strict "is this actually Azure Database for PostgreSQL" check
+    (e.g. to decide PgBouncer-aware pooling), use `is_azure_postgres_host`
+    instead."""
     if any(host.endswith(suffix) for suffix in _LAKEBASE_HOST_SUFFIXES):
         return "databricks_lakebase"
     return "azure_postgres"
 
 
-def get_azure_postgres_password() -> str:
+def is_azure_postgres_host(host: str) -> bool:
+    """True only for actual Azure Database for PostgreSQL hostnames —
+    unlike `detect_provider`, this is not a fallback default."""
+    return any(host.endswith(suffix) for suffix in _AZURE_POSTGRES_HOST_SUFFIXES)
+
+
+def get_azure_postgres_password(
+    *,
+    managed_identity: bool = False,
+    exclude_interactive_browser_credential: bool = True,
+) -> str:
+    """Fetch an Entra ID token to use as an Azure Postgres password.
+
+    `managed_identity=True` uses `ManagedIdentityCredential` (for workloads
+    running under an Azure-assigned identity); otherwise
+    `DefaultAzureCredential`, whose credential chain already falls back to
+    managed identity when no other credential is available. Both credential
+    objects are process-cached so repeated calls (one per new pooled
+    connection) don't re-probe the credential chain every time."""
     try:
-        from azure.identity import DefaultAzureCredential
+        from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
     except ImportError:
         raise ImportError("Install azure-identity extra: pip install pgdevkit[azure]")
-    cred = DefaultAzureCredential()
-    token = cred.get_token("https://ossrdbms-aad.database.windows.net/.default")
+    global _default_credential, _managed_identity_credential
+    if managed_identity:
+        if _managed_identity_credential is None:
+            _managed_identity_credential = ManagedIdentityCredential()
+        credential = _managed_identity_credential
+    else:
+        if _default_credential is None:
+            _default_credential = DefaultAzureCredential(
+                exclude_interactive_browser_credential=exclude_interactive_browser_credential
+            )
+        credential = _default_credential
+    token = credential.get_token("https://ossrdbms-aad.database.windows.net/.default")
     return token.token
 
 
@@ -33,6 +73,8 @@ def build_conninfo(
     *,
     databricks_workspace_host: str | None = None,
     databricks_instance: str | None = None,
+    managed_identity: bool = False,
+    exclude_interactive_browser_credential: bool = True,
 ) -> str:
     if entra_user is None:
         return url
@@ -50,7 +92,10 @@ def build_conninfo(
 
         password = get_lakebase_password(databricks_workspace_host, databricks_instance)
     else:
-        password = get_azure_postgres_password()
+        password = get_azure_postgres_password(
+            managed_identity=managed_identity,
+            exclude_interactive_browser_credential=exclude_interactive_browser_credential,
+        )
 
     netloc = f"{quote(entra_user, safe='')}:{quote(password, safe='')}@{host}{port}"
     return urlunparse(parsed._replace(netloc=netloc))
diff --git a/pgdevkit/db/__init__.py b/pgdevkit/db/__init__.py
index d66a75d..d8ae8b0 100644
--- a/pgdevkit/db/__init__.py
+++ b/pgdevkit/db/__init__.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from .complex_types import ComplexHelper
 from .connection import PgPool
 from .crud import (
     pg_delete,
@@ -19,6 +20,7 @@
 from .model import PostgresTableModel
 
 __all__ = [
+    "ComplexHelper",
     "PgPool",
     "PostgresTableModel",
     "SqlLoader",
diff --git a/skills/pgdevkit/references/complex_helper.py b/pgdevkit/db/complex_types.py
similarity index 65%
rename from skills/pgdevkit/references/complex_helper.py
rename to pgdevkit/db/complex_types.py
index 9fe1a39..796d5d5 100644
--- a/skills/pgdevkit/references/complex_helper.py
+++ b/pgdevkit/db/complex_types.py
@@ -1,51 +1,53 @@
-"""
-ComplexHelper — psycopg adapter for PostgreSQL composite types, enums, and JSONB.
+from __future__ import annotations
 
-Use this when your project has custom PostgreSQL types (composite types, enums)
-that need to be registered with psycopg before inserting test data. Plain columns
-and JSONB are handled automatically by pgdevkit.testdb.schema's _insert_test_data;
-you only need this class for USER-DEFINED types, which that function does not
-yet support natively — wrap the connection it's given before the INSERT:
-
-    from pgdevkit.testdb import schema  # for reference, not a public extension point yet
-    from your_project.pg_complex_helper import ComplexHelper
-
-    async def insert_test_data_with_complex_types(json_file, table, force_reset, con):
-        helper = ComplexHelper(con)
-        schema_name, table_name = table.split(".")
-        complex_types = await helper.load_all_complex_types((schema_name, table_name))
-        rows = json.loads(json_file.read_text(encoding="utf-8"))
-        for row in rows:
-            for col, info in complex_types.items():
-                if col in row:
-                    row[col] = await helper.recursive_convert(row[col], info, con)
-        # then insert `rows` the same way schema.py's _insert_test_data does
-"""
+from typing import Any, Callable
 
 from psycopg import AsyncConnection
 from psycopg.rows import dict_row
-from typing import Any
 from psycopg.sql import Identifier
 from psycopg.types.composite import CompositeInfo, register_composite
-from psycopg.types.json import Jsonb
 from psycopg.types.enum import EnumInfo, register_enum
+from psycopg.types.json import Jsonb
+
+ComplexTypeInfo = CompositeInfo | EnumInfo | type[Jsonb] | None
 
 
 class ComplexHelper:
-    complex_types: dict[tuple[str, str], CompositeInfo | EnumInfo] = {}
+    """psycopg adapter for PostgreSQL composite types, enums, and JSONB.
 
-    def __init__(self, con: AsyncConnection):
-        self.con = con
-        self.system_complex_type_dict = None
+    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.
 
+    `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.
+    """
+
+    def __init__(
+        self,
+        con: AsyncConnection,
+        normalizers: dict[str, Callable[[dict], dict]] | None = None,
+    ) -> None:
+        self.con = con
+        self.system_complex_type_dict: dict[Any, tuple[str, str]] | None = None
+        # Instance-scoped: a CompositeInfo/EnumInfo carries OIDs from a
+        # specific connection/database, so caching it on the class (shared
+        # across every connection) would leak stale OIDs across databases
+        # that happen to reuse the same type name — exactly the case for
+        # pgdevkit's per-worktree isolated test databases.
+        self.complex_types: dict[tuple[str, str], CompositeInfo | EnumInfo] = {}
         self.registered: set[CompositeInfo | EnumInfo] = set()
+        self._normalizers = normalizers or {}
 
-    async def load_complex_type_dict(self):
+    async def load_complex_type_dict(self) -> None:
         async with self.con.cursor(row_factory=dict_row) as cur:
             await cur.execute("""
         SELECT t.oid,
                 pg_catalog.format_type ( t.oid, NULL ) AS obj_name,
-				t.typtype
+                t.typtype
             FROM pg_catalog.pg_type t
             JOIN pg_catalog.pg_namespace n
                 ON n.oid = t.typnamespace
@@ -57,13 +59,9 @@ async def load_complex_type_dict(self):
                 AND n.nspname <> 'information_schema'
                 AND n.nspname !~ '^pg_toast'""")
             system_complex_types = await cur.fetchall()
-            self.system_complex_type_dict = {
-                r["oid"]: (r["obj_name"], r["typtype"]) for r in system_complex_types
-            }
+            self.system_complex_type_dict = {r["oid"]: (r["obj_name"], r["typtype"]) for r in system_complex_types}
 
-    async def _load_complex_type_from_colinfos(
-        self, res: dict[str, Any] | None
-    ) -> CompositeInfo | EnumInfo | type[Jsonb] | None:
+    async def _load_complex_type_from_colinfos(self, res: dict[str, Any] | None) -> ComplexTypeInfo:
         if not res:
             return None
         if res["data_type"].lower() == "jsonb":
@@ -78,15 +76,13 @@ async def _load_complex_type_from_colinfos(
             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
-        )
+        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
 
     async def load_all_complex_types(
         self, table_name: tuple[str, str], include_generated: bool = False
-    ) -> dict[str, CompositeInfo | type[Jsonb] | EnumInfo | None]:
+    ) -> dict[str, ComplexTypeInfo]:
         if self.system_complex_type_dict is None:
             await self.load_complex_type_dict()
         colquery = """
@@ -100,7 +96,8 @@ async def load_all_complex_types(
             udt_schema, udt_name,
             e.enum_name is not null as is_enum
               from information_schema.columns c
-                 left join enum_types e on e.enum_schema=c.udt_schema and e.enum_name=c.udt_name
+                 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)"""
         async with self.con.cursor(row_factory=dict_row) as cur:
             await cur.execute(
@@ -112,14 +109,9 @@ async def load_all_complex_types(
                 },
             )
             res = await cur.fetchall()
-            return {
-                r["column_name"]: await self._load_complex_type_from_colinfos(r)
-                for r in res
-            }
+            return {r["column_name"]: await self._load_complex_type_from_colinfos(r) for r in res}
 
-    async def load_complex_type(
-        self, table_name: tuple[str, str], col_name: str
-    ) -> CompositeInfo | type[Jsonb] | EnumInfo | None:
+    async def load_complex_type(self, table_name: tuple[str, str], col_name: str) -> ComplexTypeInfo:
         if self.system_complex_type_dict is None:
             await self.load_complex_type_dict()
         colquery = """
@@ -133,7 +125,8 @@ async def load_complex_type(
             udt_schema, udt_name,
             e.enum_name is not null as is_enum
               from information_schema.columns c
-            left join enum_types e on e.enum_schema=c.udt_schema and e.enum_name=c.udt_name
+            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 column_name = %(col)s"""
         async with self.con.cursor(row_factory=dict_row) as cur:
@@ -142,18 +135,13 @@ async def load_complex_type(
                 {"schema": table_name[0], "tbl": table_name[1], "col": col_name},
             )
             res = await cur.fetchone()
-
             return await self._load_complex_type_from_colinfos(res)
 
-    async def _get_complex_type(
-        self, name: str, is_enum: bool, con: AsyncConnection
-    ) -> CompositeInfo | EnumInfo:
+    async def _get_complex_type(self, name: str, is_enum: bool, con: AsyncConnection) -> CompositeInfo | EnumInfo:
         if name.endswith("[]"):
             name = name[:-2]
-        schema, type_name = name.split(".")
-        if type_name.startswith(
-            "_"
-        ):  # the array type in PostgreSQL starts with an underscore
+        schema, type_name = name.split(".") if "." in name else ("public", name)
+        if type_name.startswith("_"):  # the array type in PostgreSQL starts with an underscore
             type_name = type_name[1:]
         if is_enum:
             ci = await EnumInfo.fetch(con, Identifier(schema, type_name))
@@ -165,12 +153,8 @@ async def _get_complex_type(
             self.complex_types[(schema, type_name)] = ci
         return self.complex_types[(schema, type_name)]
 
-    async def _recurse_register(
-        self, info: CompositeInfo | EnumInfo, con: AsyncConnection
-    ):
-        assert self.system_complex_type_dict is not None, (
-            "System complex type dictionary not loaded"
-        )
+    async def _recurse_register(self, info: CompositeInfo | EnumInfo, con: AsyncConnection) -> 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)
@@ -188,7 +172,7 @@ async def _recurse_register(
     async def recursive_convert(
         self,
         value: Any,
-        info: CompositeInfo | EnumInfo | type[Jsonb] | None,
+        info: ComplexTypeInfo,
         con: AsyncConnection,
     ) -> Any:
         if info is None:
@@ -197,20 +181,21 @@ async def recursive_convert(
             return None
         if self.system_complex_type_dict is None:
             await self.load_complex_type_dict()
+        if info == Jsonb:
+            # A JSONB column's value is wrapped whole, even if it's a list —
+            # only an array-of-composite/enum column recurses per element.
+            return Jsonb(value)
         if isinstance(value, list):
             return [await self.recursive_convert(item, info, con) for item in value]
         prms = {}
-        if info == Jsonb:
-            return Jsonb(value)
         if isinstance(value, str):
             assert isinstance(info, EnumInfo), f"Expected EnumInfo, got {type(info)}"
             return getattr(info.enum, value)  # Enum
-        assert isinstance(info, CompositeInfo), (
-            f"Expected CompositeInfo, got {type(info)}"
-        )
-        assert self.system_complex_type_dict is not None, (
-            "System complex type dictionary not loaded"
-        )
+        assert isinstance(info, CompositeInfo), f"Expected CompositeInfo, got {type(info)}"
+        assert self.system_complex_type_dict is not None, "System complex type dictionary not loaded"
+        normalizer = self._normalizers.get(info.name)
+        if normalizer is not None:
+            value = normalizer(value)
         for k, v in value.items():
             if v is None:
                 prms[k] = None
@@ -221,14 +206,10 @@ async def recursive_convert(
                 name, typtype = self.system_complex_type_dict[type_oid]
                 ci = await self._get_complex_type(name, typtype == "e", con)
                 if name.endswith("[]"):
-                    prms[k] = [
-                        await self.recursive_convert(item, ci, con) for item in v
-                    ]
+                    prms[k] = [await self.recursive_convert(item, ci, con) for item in v]
                 else:
                     prms[k] = await self.recursive_convert(v, ci, con)
             else:
                 prms[k] = v
-        assert info.python_type is not None, (
-            f"Python type for {info.name} is null, maybe an array?"
-        )
+        assert info.python_type is not None, f"Python type for {info.name} is null, maybe an array?"
         return info.python_type(**prms) if prms else None
diff --git a/pgdevkit/db/connection.py b/pgdevkit/db/connection.py
index a2b104b..533c88a 100644
--- a/pgdevkit/db/connection.py
+++ b/pgdevkit/db/connection.py
@@ -2,10 +2,11 @@
 
 import asyncio
 import os
+from typing import Literal
 
-from psycopg_pool import AsyncConnectionPool
+from psycopg_pool import AsyncConnectionPool, AsyncNullConnectionPool
 
-from ..connection import detect_provider, get_azure_postgres_password
+from ..connection import detect_provider, get_azure_postgres_password, is_azure_postgres_host
 from ..lakebase import get_lakebase_password
 
 
@@ -18,6 +19,13 @@ class PgPool:
     token flow or Databricks Lakebase credential exchange; for the latter
     also set `{env_prefix}DATABRICKS_WORKSPACE_HOST` and
     `{env_prefix}DATABRICKS_INSTANCE`.
+
+    `use_null_pool="auto"` (the default) switches to a null pool — no local
+    pooling — when the host is Azure Postgres, since Azure's own PgBouncer
+    (transaction-mode) does the pooling; `connection_kwargs` then
+    auto-includes `prepare_threshold=None` too, since PgBouncer transaction
+    mode doesn't support server-side prepared statements. Pass an explicit
+    `use_null_pool`/`connection_kwargs` to override either.
     """
 
     def __init__(
@@ -26,11 +34,26 @@ def __init__(
         max_size: int = 40,
         *,
         entra_user: str | None = None,
+        credential_kind: Literal["default_azure", "managed_identity"] = "default_azure",
+        exclude_interactive_browser_credential: bool = True,
+        max_lifetime: float | None = None,
+        use_null_pool: bool | Literal["auto"] = "auto",
+        connection_kwargs: dict | None = None,
+        dsn_params: dict[str, str] | None = None,
     ) -> None:
         self._env_prefix = env_prefix
         self._max_size = max_size
         self._entra_user = entra_user
-        self._pool: AsyncConnectionPool | None = None
+        self._credential_kind = credential_kind
+        self._exclude_interactive_browser_credential = exclude_interactive_browser_credential
+        self._max_lifetime = max_lifetime
+        self._use_null_pool = use_null_pool
+        self._connection_kwargs = connection_kwargs
+        self._dsn_params = dsn_params or {}
+        self._pool: AsyncConnectionPool | AsyncNullConnectionPool | None = None
+
+    def _is_azure_postgres(self, host: str) -> bool:
+        return is_azure_postgres_host(host)
 
     async def _dsn(self) -> str:
         p = self._env_prefix
@@ -48,18 +71,45 @@ async def _dsn(self) -> str:
                 instance_name = os.environ[p + "DATABRICKS_INSTANCE"]
                 password = await asyncio.to_thread(get_lakebase_password, workspace_host, instance_name)
             else:
-                password = await asyncio.to_thread(get_azure_postgres_password)
+                password = await asyncio.to_thread(
+                    get_azure_postgres_password,
+                    managed_identity=self._credential_kind == "managed_identity",
+                    exclude_interactive_browser_credential=self._exclude_interactive_browser_credential,
+                )
 
-        return f"host={host} port={port} dbname={dbname} user={user} password={password}"
+        dsn = f"host={host} port={port} dbname={dbname} user={user} password={password}"
+        for key, value in self._dsn_params.items():
+            dsn += f" {key}={value}"
+        return dsn
 
     async def open(self) -> None:
         if self._pool is None:
-            self._pool = AsyncConnectionPool(
-                conninfo=self._dsn,
-                open=False,
-                max_size=self._max_size,
-                check=AsyncConnectionPool.check_connection,
-            )
+            host = os.environ[self._env_prefix + "HOST"]
+            is_azure_postgres = self._is_azure_postgres(host)
+            use_null_pool = is_azure_postgres if self._use_null_pool == "auto" else self._use_null_pool
+            connection_kwargs = self._connection_kwargs
+            if connection_kwargs is None:
+                # PgBouncer (transaction mode) doesn't support prepared statements
+                connection_kwargs = {"prepare_threshold": None} if is_azure_postgres else {}
+            max_lifetime = 3600.0 if self._max_lifetime is None else self._max_lifetime
+            if use_null_pool:
+                self._pool = AsyncNullConnectionPool(
+                    conninfo=self._dsn,
+                    open=False,
+                    max_size=self._max_size,
+                    max_lifetime=max_lifetime,
+                    check=AsyncNullConnectionPool.check_connection,
+                    kwargs=connection_kwargs,
+                )
+            else:
+                self._pool = AsyncConnectionPool(
+                    conninfo=self._dsn,
+                    open=False,
+                    max_size=self._max_size,
+                    max_lifetime=max_lifetime,
+                    check=AsyncConnectionPool.check_connection,
+                    kwargs=connection_kwargs,
+                )
         if not self._pool._opened:
             await self._pool.open()
 
@@ -72,3 +122,12 @@ def connection(self):
         if self._pool is None:
             raise RuntimeError("Call open() first (e.g. in app startup).")
         return self._pool.connection()
+
+    @property
+    def raw_pool(self) -> AsyncConnectionPool | AsyncNullConnectionPool:
+        """The underlying psycopg_pool object — for integrating with other
+        pool-consuming libraries (e.g. Procrastinate's `open_async(pool=...)`)
+        that need it directly rather than going through `connection()`."""
+        if self._pool is None:
+            raise RuntimeError("Call open() first (e.g. in app startup).")
+        return self._pool
diff --git a/pgdevkit/db/crud.py b/pgdevkit/db/crud.py
index 87ebc40..f8f05fa 100644
--- a/pgdevkit/db/crud.py
+++ b/pgdevkit/db/crud.py
@@ -4,19 +4,98 @@
 
 from psycopg.connection_async import AsyncConnection
 from psycopg.rows import dict_row
-from psycopg.sql import SQL, Identifier, Placeholder
+from psycopg.sql import SQL, Composable, Identifier, Placeholder
+from pydantic import BaseModel
 
+from .complex_types import ComplexHelper
 from .model import PostgresTableModel
 
 T = TypeVar("T", bound=PostgresTableModel)
 
 
-async def pg_retrieve(con: AsyncConnection, data_type: Type[T], pks: dict) -> T | None:
-    """Fetch a single row by primary key(s)."""
+async def _select_list(
+    con: AsyncConnection, table_name: tuple[str, str], complex_helper: ComplexHelper | None
+) -> Composable:
+    """Column list for a SELECT, wrapping composite/enum/JSONB columns in
+    `to_jsonb(...)` so psycopg gets back plain Python values. Falls back to
+    `SELECT *` (no extra query) when no ComplexHelper is given."""
+    if complex_helper is None:
+        return SQL("*")
+    complex_types = await complex_helper.load_all_complex_types(table_name, include_generated=True)
+    if not complex_types:
+        return SQL("*")
+    parts = [
+        SQL("to_jsonb({col}) as {col}").format(col=Identifier(col)) if info is not None else Identifier(col)
+        for col, info in complex_types.items()
+    ]
+    return SQL(", ").join(parts)
+
+
+async def _convert_complex_values(
+    con: AsyncConnection,
+    table_name: tuple[str, str],
+    data: dict,
+    complex_helper: ComplexHelper | None,
+) -> dict:
+    """Convert dict/list values destined for composite/enum columns into the
+    psycopg-registered types those columns need. A no-op (no extra query)
+    unless `data` actually contains dict/list values."""
+    if complex_helper is None:
+        return data
+    candidate_keys = [k for k, v in data.items() if isinstance(v, (dict, list))]
+    if not candidate_keys:
+        return data
+    converted = dict(data)
+    for k in candidate_keys:
+        info = await complex_helper.load_complex_type(table_name, k)
+        if info is not None:
+            converted[k] = await complex_helper.recursive_convert(data[k], info, con)
+    return converted
+
+
+async def _convert_complex_values_many(
+    con: AsyncConnection,
+    table_name: tuple[str, str],
+    rows: Sequence[dict],
+    complex_helper: ComplexHelper | None,
+) -> Sequence[dict]:
+    if complex_helper is None or not rows:
+        return rows
+    candidate_keys = {k for row in rows for k, v in row.items() if isinstance(v, (dict, list))}
+    if not candidate_keys:
+        return rows
+    infos = {k: await complex_helper.load_complex_type(table_name, k) for k in candidate_keys}
+    complex_keys = {k for k, info in infos.items() if info is not None}
+    if not complex_keys:
+        return rows
+    converted_rows = []
+    for row in rows:
+        new_row = dict(row)
+        for k in complex_keys:
+            if isinstance(new_row.get(k), (dict, list)):
+                new_row[k] = await complex_helper.recursive_convert(new_row[k], infos[k], con)
+        converted_rows.append(new_row)
+    return converted_rows
+
+
+async def pg_retrieve(
+    con: AsyncConnection,
+    data_type: Type[T],
+    pks: dict,
+    *,
+    complex_helper: ComplexHelper | None = None,
+) -> T | None:
+    """Fetch a single row by primary key(s).
+
+    Pass `complex_helper` (a `ComplexHelper`, optionally configured with
+    `normalizers`) when the table has composite/enum columns; omitted, this
+    behaves exactly like a plain `SELECT *`."""
     async with con.cursor(row_factory=dict_row) as cur:
-        schema, table = data_type.get_table_name()
-        query = SQL("SELECT * FROM {tbl} WHERE {where}").format(
-            tbl=Identifier(schema, table),
+        table_name = data_type.get_table_name()
+        select_cols = await _select_list(con, table_name, complex_helper)
+        query = SQL("SELECT {cols} FROM {tbl} WHERE {where}").format(
+            cols=select_cols,
+            tbl=Identifier(*table_name),
             where=SQL(" AND ").join(SQL("{col} = {val}").format(col=Identifier(pk), val=Placeholder(pk)) for pk in pks),
         )
         await cur.execute(query, pks)
@@ -30,27 +109,37 @@ async def pg_retrieve_many(
     filters: dict,
     *,
     from_dict: Optional[Callable[[Mapping], T]] = None,
+    complex_helper: ComplexHelper | None = None,
 ) -> Sequence[T]:
     """Fetch multiple rows matching all filter key=value pairs."""
     async with con.cursor(row_factory=dict_row) as cur:
-        schema, table = data_type.get_table_name()
+        table_name = data_type.get_table_name()
+        select_cols = await _select_list(con, table_name, complex_helper)
         if filters:
-            query = SQL("SELECT * FROM {tbl} WHERE {where}").format(
-                tbl=Identifier(schema, table),
+            query = SQL("SELECT {cols} FROM {tbl} WHERE {where}").format(
+                cols=select_cols,
+                tbl=Identifier(*table_name),
                 where=SQL(" AND ").join(
                     SQL("{col} = {val}").format(col=Identifier(k), val=Placeholder(k)) for k in filters
                 ),
             )
         else:
-            query = SQL("SELECT * FROM {tbl}").format(tbl=Identifier(schema, table))
+            query = SQL("SELECT {cols} FROM {tbl}").format(cols=select_cols, tbl=Identifier(*table_name))
         await cur.execute(query, filters)
         rows = await cur.fetchall()
     fn = from_dict or (lambda d: data_type(**d))
     return [fn(r) for r in rows]
 
 
-async def pg_insert(con: AsyncConnection, table_name: tuple[str, str], data: dict) -> dict[str, Any]:
+async def pg_insert(
+    con: AsyncConnection,
+    table_name: tuple[str, str],
+    data: dict,
+    *,
+    complex_helper: ComplexHelper | None = None,
+) -> dict[str, Any]:
     """Insert one row and return the full row (RETURNING *)."""
+    data = await _convert_complex_values(con, table_name, data, complex_helper)
     query = SQL("INSERT INTO {tbl} ({cols}) VALUES ({vals}) RETURNING *").format(
         tbl=Identifier(*table_name),
         cols=SQL(", ").join(Identifier(k) for k in data),
@@ -94,8 +183,11 @@ async def pg_upsert_dict(
     table_name: tuple[str, str],
     data: dict,
     primary_keys: Sequence[str],
+    *,
+    complex_helper: ComplexHelper | None = None,
 ) -> dict:
     """INSERT ... ON CONFLICT ... DO UPDATE, returns the row as a dict."""
+    data = await _convert_complex_values(con, table_name, data, complex_helper)
     fields = list(data)
     updates = [SQL("{col} = EXCLUDED.{col}").format(col=Identifier(k)) for k in fields]
     query = SQL(
@@ -114,9 +206,13 @@ async def pg_upsert_dict(
     return row
 
 
-async def pg_upsert(con: AsyncConnection, data: T, data_type: type[T]) -> dict:
+async def pg_upsert(
+    con: AsyncConnection, data: T, data_type: type[T], *, complex_helper: ComplexHelper | None = None
+) -> dict:
     """Upsert a typed model instance."""
-    return await pg_upsert_dict(con, data_type.get_table_name(), data.model_dump(), data_type.get_primary_key())
+    return await pg_upsert_dict(
+        con, data_type.get_table_name(), data.model_dump(), data_type.get_primary_key(), complex_helper=complex_helper
+    )
 
 
 async def pg_upsert_many_dict(
@@ -124,45 +220,76 @@ async def pg_upsert_many_dict(
     table_name: tuple[str, str],
     data: Sequence[dict],
     primary_keys: Sequence[str],
+    *,
+    must_exist: bool = False,
+    complex_helper: ComplexHelper | None = None,
 ) -> None:
-    """Batch upsert — one round-trip via executemany."""
+    """Batch upsert — one round-trip via executemany.
+
+    `must_exist=True` switches to a plain UPDATE (no INSERT) matched on
+    `primary_keys` — for callers that only ever update pre-existing rows and
+    want a missing row to be a silent no-op rather than create one."""
     if not data:
         return
+    data = await _convert_complex_values_many(con, table_name, data, complex_helper)
     fields = list(data[0])
-    updates = [SQL("{col} = EXCLUDED.{col}").format(col=Identifier(k)) for k in fields if k not in primary_keys]
-    query = SQL("INSERT INTO {tbl} ({cols}) VALUES ({vals}) ON CONFLICT ({pks}) DO UPDATE SET {updates}").format(
-        tbl=Identifier(*table_name),
-        cols=SQL(", ").join(Identifier(k) for k in fields),
-        vals=SQL(", ").join(Placeholder(k) for k in fields),
-        pks=SQL(", ").join(Identifier(pk) for pk in primary_keys),
-        updates=SQL(", ").join(updates),
-    )
+    if must_exist:
+        update_assignments = [
+            SQL("{col} = {val}").format(col=Identifier(k), val=Placeholder(k)) for k in fields if k not in primary_keys
+        ]
+        target_eq = SQL(" AND ").join(
+            SQL("t.{col} = {val}").format(col=Identifier(pk), val=Placeholder(pk)) for pk in primary_keys
+        )
+        query = SQL("UPDATE {tbl} t SET {updates} WHERE {target_eq}").format(
+            tbl=Identifier(*table_name),
+            updates=SQL(", ").join(update_assignments),
+            target_eq=target_eq,
+        )
+    else:
+        updates = [SQL("{col} = EXCLUDED.{col}").format(col=Identifier(k)) for k in fields if k not in primary_keys]
+        query = SQL("INSERT INTO {tbl} ({cols}) VALUES ({vals}) ON CONFLICT ({pks}) DO UPDATE SET {updates}").format(
+            tbl=Identifier(*table_name),
+            cols=SQL(", ").join(Identifier(k) for k in fields),
+            vals=SQL(", ").join(Placeholder(k) for k in fields),
+            pks=SQL(", ").join(Identifier(pk) for pk in primary_keys),
+            updates=SQL(", ").join(updates),
+        )
     async with con.cursor() as cur:
         await cur.executemany(query, data)
 
 
-async def pg_upsert_many(con: AsyncConnection, data: Sequence[T], data_type: type[T]) -> None:
+async def pg_upsert_many(
+    con: AsyncConnection, data: Sequence[T], data_type: type[T], *, complex_helper: ComplexHelper | None = None
+) -> None:
     await pg_upsert_many_dict(
-        con, data_type.get_table_name(), [d.model_dump() for d in data], data_type.get_primary_key()
+        con,
+        data_type.get_table_name(),
+        [d.model_dump() for d in data],
+        data_type.get_primary_key(),
+        complex_helper=complex_helper,
     )
 
 
 async def pg_insert_many(
     con: AsyncConnection,
     table_name: tuple[str, str],
-    data: Sequence[dict],
+    data: Sequence[dict | BaseModel],
+    *,
+    complex_helper: ComplexHelper | None = None,
 ) -> None:
     """Batch insert — no RETURNING, one round-trip via executemany."""
     if not data:
         return
-    fields = list(data[0])
+    dict_data = [d if isinstance(d, dict) else d.model_dump() for d in data]
+    dict_data = await _convert_complex_values_many(con, table_name, dict_data, complex_helper)
+    fields = list(dict_data[0])
     query = SQL("INSERT INTO {tbl} ({cols}) VALUES ({vals})").format(
         tbl=Identifier(*table_name),
         cols=SQL(", ").join(Identifier(k) for k in fields),
         vals=SQL(", ").join(Placeholder(k) for k in fields),
     )
     async with con.cursor() as cur:
-        await cur.executemany(query, data)
+        await cur.executemany(query, dict_data)
 
 
 async def pg_delete_dict(con: AsyncConnection, table_name: tuple[str, str], data: dict) -> dict | None:
diff --git a/pgdevkit/testdb/constants.py b/pgdevkit/testdb/constants.py
index 841552e..09b39ca 100644
--- a/pgdevkit/testdb/constants.py
+++ b/pgdevkit/testdb/constants.py
@@ -20,9 +20,9 @@ def conninfo(dbname: str, *, connect_timeout: int | None = None) -> str:
     be pointed at a unix socket directory (e.g. /var/run/postgresql) and
     authenticate via peer auth as the current OS user instead of a password.
     """
-    params: dict[str, str | int] = {"host": HOST, "port": PORT, "user": USER, "dbname": dbname}
+    params: dict[str, str] = {"host": HOST, "port": str(PORT), "user": USER, "dbname": dbname}
     if PASSWORD:
         params["password"] = PASSWORD
     if connect_timeout is not None:
-        params["connect_timeout"] = connect_timeout
+        params["connect_timeout"] = str(connect_timeout)
     return make_conninfo(**params)
diff --git a/pgdevkit/testdb/schema.py b/pgdevkit/testdb/schema.py
index 77fddd8..9633ea2 100644
--- a/pgdevkit/testdb/schema.py
+++ b/pgdevkit/testdb/schema.py
@@ -13,6 +13,8 @@
 from psycopg.rows import dict_row
 from psycopg.sql import SQL, Identifier, Placeholder
 
+from ..db.complex_types import ComplexHelper
+
 logger = logging.getLogger(__name__)
 logging.getLogger("sqlglot").setLevel(logging.ERROR)
 
@@ -126,7 +128,11 @@ def _iter_sql_files(database_dir: Path):
 
 
 async def _insert_test_data(
-    json_file: Path, table: str, force_reset: bool, con: psycopg.AsyncConnection
+    json_file: Path,
+    table: str,
+    force_reset: bool,
+    con: psycopg.AsyncConnection,
+    complex_helper: ComplexHelper,
 ) -> None:
     if not json_file.exists():
         return
@@ -142,11 +148,20 @@ async def _insert_test_data(
             if row and row["cnt"] == len(rows):
                 return
 
-        col_names = list(rows[0].keys())
+        complex_types = await complex_helper.load_all_complex_types((schema, table_name))
+        # Silently drop JSON keys that don't correspond to a real column —
+        # e.g. a stale fixture left over from a since-renamed/removed column.
+        col_names = [c for c in rows[0] if c in complex_types]
         for row in rows:
             for col in col_names:
-                if isinstance(row[col], (dict, list)):
-                    row[col] = json.dumps(row[col])
+                info = complex_types.get(col)
+                if info is not None and isinstance(row[col], (dict, list)):
+                    # composite/enum/JSONB: needs psycopg-registered-type
+                    # conversion. Anything else (plain scalars, and native
+                    # Postgres arrays like text[], which aren't "complex" —
+                    # psycopg already adapts a Python list to those natively)
+                    # is left untouched.
+                    row[col] = await complex_helper.recursive_convert(row[col], info, con)
 
         await cur.execute(SQL("DELETE FROM {t}").format(t=Identifier(schema, table_name)))
         insert_sql = SQL("INSERT INTO {t} ({cols}) VALUES ({vals})").format(
@@ -157,24 +172,66 @@ async def _insert_test_data(
         await cur.executemany(insert_sql, rows)
 
 
+_UNSAFE_MIGRATION_PATTERNS = (
+    "DROP COLUMN",
+    "DROP TABLE",
+    "TRUNCATE",
+    "RENAME COLUMN",
+    "RENAME TO",
+    "DELETE FROM",
+    "OWNER TO",
+)
+
+
+def _is_additive_migration(sql: str) -> bool:
+    """Only pure-additive migrations (ADD COLUMN/CREATE ... IF NOT EXISTS,
+    etc.) are safe to (re-)apply against a schema that already reflects
+    later migrations — skip anything that could drop, rename, truncate, or
+    delete existing structure/data, alter an existing column, or change
+    ownership. This is a substring heuristic, not a SQL parser, so it errs
+    towards skipping (a false negative here means real data loss replayed
+    on every apply_schema() call) rather than trying to be exhaustive."""
+    normalized = sql.upper()
+    has_alter_column = "ALTER COLUMN" in normalized and "ADD COLUMN" not in normalized
+    has_unsafe_pattern = any(pattern in normalized for pattern in _UNSAFE_MIGRATION_PATTERNS)
+    return not (has_alter_column or has_unsafe_pattern)
+
+
+def _iter_migration_files(database_dir: Path):
+    """Yield every `*.sql` file in any `migrations/` subdirectory, in
+    filename order (the project's naming convention — numeric or date
+    prefixes — sorts chronologically)."""
+    for root, _, files in os.walk(database_dir):
+        if Path(root).name != "migrations":
+            continue
+        for file in sorted(files):
+            if file.endswith(".sql"):
+                yield Path(root) / file
+
+
 async def apply_schema(
     con: psycopg.AsyncConnection,
     database_dir: Path,
     extensions: tuple[str, ...] = (),
     force_reset: bool = False,
 ) -> None:
-    """Apply every .sql file under database_dir (in dependency-safe order)
-    and seed any matching .test_data.json files. Safe to call repeatedly."""
+    """Apply every .sql file under database_dir (in dependency-safe order),
+    seed any matching .test_data.json files, then apply purely-additive
+    `migrations/*.sql` files (see `_is_additive_migration`) so a schema that
+    ships changes via migration files rather than editing the base object
+    files stays in sync. Safe to call repeatedly."""
     await con.set_autocommit(True)
     for extension in extensions:
         await con.execute(SQL("CREATE EXTENSION IF NOT EXISTS {e}").format(e=Identifier(extension)))
 
+    complex_helper = ComplexHelper(con)
+
     async def _apply(file: Path, sql: str) -> None:
         await con.execute(cast(Any, sql))
         json_file = file.with_suffix(".test_data.json")
         if json_file.exists():
             schema_name = _strip_layer_prefix(file.parent.parent.name)
-            await _insert_test_data(json_file, f"{schema_name}.{file.stem}", force_reset, con)
+            await _insert_test_data(json_file, f"{schema_name}.{file.stem}", force_reset, con, complex_helper)
 
     failures: list[tuple[Path, str]] = []
     for file, sql in _iter_sql_files(database_dir):
@@ -186,3 +243,13 @@ async def _apply(file: Path, sql: str) -> None:
 
     for file, sql in failures:
         await _apply(file, sql)
+
+    for migration_file in _iter_migration_files(database_dir):
+        content = migration_file.read_text(encoding="utf-8")
+        if not _is_additive_migration(content):
+            continue
+        try:
+            await con.execute(cast(Any, content))
+            logger.info("Applied migration %s", migration_file.name)
+        except Exception as e:  # noqa: BLE001
+            logger.debug("Migration %s skipped (likely already applied): %s", migration_file.name, e)
diff --git a/pyproject.toml b/pyproject.toml
index 20219e2..8e3e4f8 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -11,7 +11,7 @@ packages = ["pgdevkit"]
 
 [project]
 name = "pgdevkit"
-version = "0.1.0"
+version = "0.2.0"
 description = "A helper for developing with Postgres"
 readme = "README.md"
 requires-python = ">=3.14"
diff --git a/skills/pgdevkit/SKILL.md b/skills/pgdevkit/SKILL.md
index b8fc97f..b785e18 100644
--- a/skills/pgdevkit/SKILL.md
+++ b/skills/pgdevkit/SKILL.md
@@ -230,9 +230,14 @@ join latest_order as lo on lo.user_id = u.id and lo.rn = 1
 
 See [`references/temporal-tables.md`](references/temporal-tables.md) for row-level history via `nearform/temporal_tables`.
 
-### Custom Postgres types in test data
-
-`pgdevkit.testdb.schema`'s test-data seeding handles plain columns and JSONB, but not composite types or enums directly. See [`references/complex_helper.py`](references/complex_helper.py) for a psycopg adapter (`ComplexHelper`) if a project needs that.
+### Custom Postgres types (composites, enums)
+
+`pgdevkit.db.complex_types.ComplexHelper` detects a table's composite/enum/JSONB
+columns and converts plain dict/list values into the psycopg-registered types
+those columns need — used automatically by both `pgdevkit.testdb.schema`'s
+test-data seeding and `pgdevkit.db.crud`'s CRUD helpers. Pass a `normalizers`
+dict (keyed by composite type name) if a project needs to reshape a value
+before conversion (e.g. backfilling missing locale keys).
 
 ### SQL formatting
 
diff --git a/tests/db/test_complex_types.py b/tests/db/test_complex_types.py
new file mode 100644
index 0000000..0e60b8f
--- /dev/null
+++ b/tests/db/test_complex_types.py
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import psycopg
+import pytest
+from psycopg.types.enum import EnumInfo
+
+from pgdevkit.db.complex_types import ComplexHelper
+from pgdevkit.db.crud import _select_list
+from pgdevkit.testdb import constants
+from pgdevkit.testdb.container import ensure_container
+from tests.testdb.conftest import RUN_SUFFIX, requires_podman
+
+TEST_DB = f"pgdevkit_complextypes_selftest_{RUN_SUFFIX}"
+
+
+def _admin_dsn() -> str:
+    return f"postgresql://{constants.USER}:{constants.PASSWORD}@{constants.HOST}:{constants.PORT}/postgres"
+
+
+def _db_dsn() -> str:
+    return f"postgresql://{constants.USER}:{constants.PASSWORD}@{constants.HOST}:{constants.PORT}/{TEST_DB}"
+
+
+@pytest.fixture
+def complex_types_test_db():
+    ensure_container()
+    with psycopg.connect(_admin_dsn(), autocommit=True) as con:
+        con.execute(f'DROP DATABASE IF EXISTS "{TEST_DB}"')
+        con.execute(f'CREATE DATABASE "{TEST_DB}"')
+    yield
+    with psycopg.connect(_admin_dsn(), autocommit=True) as con:
+        con.execute(f'DROP DATABASE IF EXISTS "{TEST_DB}"')
+
+
+@requires_podman
+async def test_array_of_enum_column_is_detected_and_converted(complex_types_test_db):
+    # Regression: information_schema.columns.udt_name for an array-of-enum
+    # column is the underscore-prefixed array type name (e.g. "_mood"), which
+    # never matched the enum_types CTE's bare enum type names — so is_enum
+    # was always computed False for such columns, and load_all_complex_types
+    # returned a CompositeInfo instead of an EnumInfo, crashing
+    # recursive_convert's `assert isinstance(info, EnumInfo)`.
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await con.execute("CREATE TYPE mood AS ENUM ('happy', 'sad')")
+        await con.execute("CREATE TABLE gadget (id serial PRIMARY KEY, moods mood[])")
+
+        helper = ComplexHelper(con)
+        types = await helper.load_all_complex_types(("public", "gadget"))
+        info = types["moods"]
+        assert isinstance(info, EnumInfo)
+
+        converted = await helper.recursive_convert(["happy", "sad"], info, con)
+        await con.execute("INSERT INTO gadget (id, moods) VALUES (1, %s)", (converted,))
+        async with con.cursor() as cur:
+            await cur.execute("SELECT moods FROM gadget WHERE id = 1")
+            (moods,) = await cur.fetchone()
+    assert [m.name for m in moods] == ["happy", "sad"]
+
+
+@requires_podman
+async def test_select_list_includes_generated_columns(complex_types_test_db):
+    # Regression: _select_list built its column list from
+    # load_all_complex_types(table_name) with the default include_generated
+    # =False, so any GENERATED ALWAYS column was silently dropped from the
+    # SELECT whenever a complex_helper was passed — unlike the plain
+    # `SELECT *` path (used when no complex_helper is given), which always
+    # includes generated columns.
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await con.execute("""
+            CREATE TABLE t (
+                id serial PRIMARY KEY,
+                price numeric,
+                qty numeric,
+                total numeric GENERATED ALWAYS AS (price * qty) STORED
+            )
+        """)
+        helper = ComplexHelper(con)
+        select = await _select_list(con, ("public", "t"), helper)
+        rendered = select.as_string(con)
+    assert '"total"' in rendered
+
+
+def test_complex_types_cache_is_per_instance_not_shared():
+    # Regression: complex_types used to be a mutable class attribute, so a
+    # CompositeInfo/EnumInfo (which carries OIDs from one specific
+    # connection/database) fetched by one ComplexHelper would leak into any
+    # other ComplexHelper that happens to look up the same type NAME —
+    # exactly what happens across pgdevkit's per-worktree isolated test
+    # databases, which legitimately reuse type names like "locale_labels"
+    # with different OIDs per database.
+    helper_a = ComplexHelper(con=None)  # con is unused by this path
+    helper_b = ComplexHelper(con=None)
+
+    helper_a.complex_types[("app", "dimensions")] = object()  # type: ignore[assignment]
+
+    assert helper_b.complex_types == {}
+    assert helper_a.complex_types is not helper_b.complex_types
diff --git a/tests/db/test_connection.py b/tests/db/test_connection.py
index 2509eea..6223d12 100644
--- a/tests/db/test_connection.py
+++ b/tests/db/test_connection.py
@@ -24,7 +24,10 @@ async def test_dsn_azure_postgres_entra(monkeypatch):
     monkeypatch.setenv(f"{ENV_PREFIX}HOST", "myserver.postgres.database.azure.com")
     monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
     monkeypatch.setenv(f"{ENV_PREFIX}DB", "mydb")
-    monkeypatch.setattr("pgdevkit.db.connection.get_azure_postgres_password", lambda: "AADTOKEN")
+    monkeypatch.setattr(
+        "pgdevkit.db.connection.get_azure_postgres_password",
+        lambda **kwargs: "AADTOKEN",
+    )
 
     pool = PgPool(env_prefix=ENV_PREFIX, entra_user="alice@example.com")
     dsn = await pool._dsn()
@@ -34,6 +37,84 @@ async def test_dsn_azure_postgres_entra(monkeypatch):
     )
 
 
+async def test_dsn_azure_postgres_entra_managed_identity(monkeypatch):
+    monkeypatch.setenv(f"{ENV_PREFIX}HOST", "myserver.postgres.database.azure.com")
+    monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
+    monkeypatch.setenv(f"{ENV_PREFIX}DB", "mydb")
+
+    calls = []
+    monkeypatch.setattr(
+        "pgdevkit.db.connection.get_azure_postgres_password",
+        lambda **kwargs: calls.append(kwargs) or "MITOKEN",
+    )
+
+    pool = PgPool(env_prefix=ENV_PREFIX, entra_user="alice@example.com", credential_kind="managed_identity")
+    dsn = await pool._dsn()
+    assert "password=MITOKEN" in dsn
+    assert calls == [{"managed_identity": True, "exclude_interactive_browser_credential": True}]
+
+
+async def test_dsn_extra_params_appended(monkeypatch):
+    monkeypatch.setenv(f"{ENV_PREFIX}HOST", "localhost")
+    monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
+    monkeypatch.setenv(f"{ENV_PREFIX}DB", "mydb")
+    monkeypatch.setenv(f"{ENV_PREFIX}USER", "myuser")
+    monkeypatch.setenv(f"{ENV_PREFIX}PASSWORD", "mypassword")
+
+    pool = PgPool(
+        env_prefix=ENV_PREFIX,
+        dsn_params={"sslmode": "require", "application_name": "myapp"},
+    )
+    dsn = await pool._dsn()
+    assert dsn == (
+        "host=localhost port=5432 dbname=mydb user=myuser password=mypassword "
+        "sslmode=require application_name=myapp"
+    )
+
+
+async def test_open_uses_null_pool_and_prepare_threshold_for_azure_host(monkeypatch):
+    monkeypatch.setenv(f"{ENV_PREFIX}HOST", "myserver.postgres.database.azure.com")
+    monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
+    monkeypatch.setenv(f"{ENV_PREFIX}DB", "mydb")
+    monkeypatch.setenv(f"{ENV_PREFIX}USER", "myuser")
+    monkeypatch.setenv(f"{ENV_PREFIX}PASSWORD", "mypassword")
+
+    pool = PgPool(env_prefix=ENV_PREFIX)
+    await pool.open()
+    try:
+        from psycopg_pool import AsyncNullConnectionPool
+
+        assert isinstance(pool._pool, AsyncNullConnectionPool)
+        assert pool._pool.kwargs == {"prepare_threshold": None}
+    finally:
+        await pool.close()
+
+
+async def test_open_uses_regular_pool_for_non_azure_host(monkeypatch):
+    monkeypatch.setenv(f"{ENV_PREFIX}HOST", "localhost")
+    monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
+    monkeypatch.setenv(f"{ENV_PREFIX}DB", "mydb")
+    monkeypatch.setenv(f"{ENV_PREFIX}USER", "myuser")
+    monkeypatch.setenv(f"{ENV_PREFIX}PASSWORD", "mypassword")
+
+    pool = PgPool(env_prefix=ENV_PREFIX)
+    await pool.open()
+    try:
+        from psycopg_pool import AsyncConnectionPool
+
+        assert type(pool._pool) is AsyncConnectionPool
+        assert pool._pool.kwargs == {}
+        assert pool.raw_pool is pool._pool
+    finally:
+        await pool.close()
+
+
+def test_raw_pool_before_open_raises():
+    pool = PgPool(env_prefix=ENV_PREFIX)
+    with pytest.raises(RuntimeError, match="Call open"):
+        pool.raw_pool
+
+
 async def test_dsn_databricks_lakebase_entra(monkeypatch):
     monkeypatch.setenv(f"{ENV_PREFIX}HOST", "instance-abc.database.azuredatabricks.net")
     monkeypatch.setenv(f"{ENV_PREFIX}PORT", "5432")
diff --git a/tests/db/test_crud.py b/tests/db/test_crud.py
index 2fdeb1a..a0992e0 100644
--- a/tests/db/test_crud.py
+++ b/tests/db/test_crud.py
@@ -7,15 +7,18 @@
 from pydantic import ConfigDict
 
 from pgdevkit.db import (
+    ComplexHelper,
     PgPool,
     PostgresTableModel,
     pg_delete,
     pg_insert,
+    pg_insert_many,
     pg_retrieve,
     pg_retrieve_many,
     pg_update,
     pg_upsert,
     pg_upsert_many,
+    pg_upsert_many_dict,
 )
 from pgdevkit.testdb import constants
 from pgdevkit.testdb.container import ensure_container
@@ -43,6 +46,20 @@ def get_primary_key() -> Sequence[str]:
         return ["id"]
 
 
+class Gizmo(PostgresTableModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: int
+    label: dict
+
+    @staticmethod
+    def get_table_name() -> tuple[str, str]:
+        return ("public", "gizmo")
+
+    @staticmethod
+    def get_primary_key() -> Sequence[str]:
+        return ["id"]
+
+
 @pytest.fixture
 async def pool(monkeypatch: pytest.MonkeyPatch):
     ensure_container()
@@ -51,6 +68,8 @@ async def pool(monkeypatch: pytest.MonkeyPatch):
         con.execute(f'CREATE DATABASE "{TEST_DB}"')
     with psycopg.connect(constants.conninfo(TEST_DB), autocommit=True) as con:
         con.execute("CREATE TABLE widget (id serial PRIMARY KEY, name text NOT NULL)")
+        con.execute("CREATE TYPE label_pair AS (en text, de text)")
+        con.execute("CREATE TABLE gizmo (id serial PRIMARY KEY, label label_pair NOT NULL)")
 
     monkeypatch.setenv(f"{ENV_PREFIX}HOST", constants.HOST)
     monkeypatch.setenv(f"{ENV_PREFIX}PORT", str(constants.PORT))
@@ -106,3 +125,62 @@ async def test_upsert_and_upsert_many(pool: PgPool):
         fetched = await pg_retrieve(con, Widget, {"id": widget2.id})
         assert fetched is not None
         assert fetched.name == "new-widget"
+
+
+@requires_podman
+async def test_upsert_many_dict_must_exist_updates_without_inserting(pool: PgPool):
+    async with pool.connection() as con:
+        inserted = await pg_insert(con, ("public", "widget"), {"name": "sprocket"})
+        existing_id = inserted["id"]
+        missing_id = existing_id + 1000
+
+        await pg_upsert_many_dict(
+            con,
+            ("public", "widget"),
+            [{"id": existing_id, "name": "updated"}, {"id": missing_id, "name": "ghost"}],
+            ["id"],
+            must_exist=True,
+        )
+
+        updated = await pg_retrieve(con, Widget, {"id": existing_id})
+        assert updated is not None
+        assert updated.name == "updated"
+        assert await pg_retrieve(con, Widget, {"id": missing_id}) is None
+
+
+@requires_podman
+async def test_insert_many_accepts_model_instances(pool: PgPool):
+    async with pool.connection() as con:
+        await pg_insert_many(
+            con,
+            ("public", "widget"),
+            [Widget(id=5001, name="from-model-a"), Widget(id=5002, name="from-model-b")],
+        )
+        fetched = await pg_retrieve_many(con, Widget, {})
+        names = {w.name for w in fetched}
+        assert {"from-model-a", "from-model-b"} <= names
+
+
+@requires_podman
+async def test_insert_retrieve_composite_column_with_normalizer(pool: PgPool):
+    # label_pair mirrors a real project's locale-labels composite type: a
+    # normalizer backfills a missing locale before the value is converted
+    # into the psycopg-registered composite type for INSERT. `RETURNING *`
+    # gives back the psycopg-generated composite instance directly (since
+    # register_composite() also wires up decoding); pg_retrieve's to_jsonb()
+    # wrapping instead unwraps it back into a plain dict.
+    def backfill_de(value: dict) -> dict:
+        if not value.get("de"):
+            value = {**value, "de": f"[{value['en']}]"}
+        return value
+
+    async with pool.connection() as con:
+        helper = ComplexHelper(con, normalizers={"label_pair": backfill_de})
+        inserted = await pg_insert(con, ("public", "gizmo"), {"label": {"en": "Hello"}}, complex_helper=helper)
+        assert inserted["label"].en == "Hello"
+        assert inserted["label"].de == "[Hello]"
+        gizmo_id = inserted["id"]
+
+        fetched = await pg_retrieve(con, Gizmo, {"id": gizmo_id}, complex_helper=helper)
+        assert fetched is not None
+        assert fetched.label == {"en": "Hello", "de": "[Hello]"}
diff --git a/tests/test_connection.py b/tests/test_connection.py
index 0e6b5bb..00315c6 100644
--- a/tests/test_connection.py
+++ b/tests/test_connection.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from pgdevkit.connection import build_conninfo, detect_provider
+from pgdevkit.connection import build_conninfo, detect_provider, is_azure_postgres_host
 
 
 @pytest.mark.parametrize(
@@ -20,13 +20,27 @@ def test_detect_provider(host, expected):
     assert detect_provider(host) == expected
 
 
+@pytest.mark.parametrize(
+    "host,expected",
+    [
+        ("myserver.postgres.database.azure.com", True),
+        ("myserver.postgres.cosmos.azure.com", True),
+        ("localhost", False),
+        ("db.internal.example.com", False),
+        ("instance-abc123.database.azuredatabricks.net", False),
+    ],
+)
+def test_is_azure_postgres_host(host, expected):
+    assert is_azure_postgres_host(host) is expected
+
+
 def test_build_conninfo_without_entra_user_returns_url_unchanged():
     url = "postgresql://user:pass@host:5432/db"
     assert build_conninfo(url) == url
 
 
 def test_build_conninfo_azure_postgres_uses_token_as_password(monkeypatch):
-    monkeypatch.setattr("pgdevkit.connection.get_azure_postgres_password", lambda: "TOKEN123")
+    monkeypatch.setattr("pgdevkit.connection.get_azure_postgres_password", lambda **kwargs: "TOKEN123")
     conninfo = build_conninfo(
         "postgresql://myserver.postgres.database.azure.com:5432/db", "alice@example.com"
     )
diff --git a/tests/testdb/fixtures/database/app/migrations/001_add_gadget_note.sql b/tests/testdb/fixtures/database/app/migrations/001_add_gadget_note.sql
new file mode 100644
index 0000000..89d1f6c
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/migrations/001_add_gadget_note.sql
@@ -0,0 +1,2 @@
+ALTER TABLE app.gadget
+    ADD COLUMN IF NOT EXISTS note text;
diff --git a/tests/testdb/fixtures/database/app/migrations/002_unsafe_drop_gadget_note.sql b/tests/testdb/fixtures/database/app/migrations/002_unsafe_drop_gadget_note.sql
new file mode 100644
index 0000000..c9f21ad
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/migrations/002_unsafe_drop_gadget_note.sql
@@ -0,0 +1,3 @@
+-- Deliberately destructive: must be skipped by apply_schema's additive-only
+-- filter, not applied. See test_apply_schema_skips_unsafe_migrations.
+ALTER TABLE app.gadget DROP COLUMN note;
diff --git a/tests/testdb/fixtures/database/app/tables/gadget.sql b/tests/testdb/fixtures/database/app/tables/gadget.sql
new file mode 100644
index 0000000..20272a0
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/tables/gadget.sql
@@ -0,0 +1,6 @@
+CREATE TABLE IF NOT EXISTS app.gadget (
+    id serial PRIMARY KEY,
+    mood app.mood NOT NULL,
+    size app.dimensions,
+    tags jsonb
+);
diff --git a/tests/testdb/fixtures/database/app/tables/gadget.test_data.json b/tests/testdb/fixtures/database/app/tables/gadget.test_data.json
new file mode 100644
index 0000000..0048dbe
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/tables/gadget.test_data.json
@@ -0,0 +1 @@
+[{"id": 1, "mood": "happy", "size": {"width": 10, "height": 20}, "tags": ["small", "shiny"], "stale_removed_column": "should be silently dropped"}]
diff --git a/tests/testdb/fixtures/database/app/types/dimensions.sql b/tests/testdb/fixtures/database/app/types/dimensions.sql
new file mode 100644
index 0000000..5bcc87b
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/types/dimensions.sql
@@ -0,0 +1,11 @@
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_type WHERE typname = 'dimensions' AND typnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'app')
+    ) THEN
+        CREATE TYPE app.dimensions AS (
+            width int,
+            height int
+        );
+    END IF;
+END$$;
diff --git a/tests/testdb/fixtures/database/app/types/mood.sql b/tests/testdb/fixtures/database/app/types/mood.sql
new file mode 100644
index 0000000..025c750
--- /dev/null
+++ b/tests/testdb/fixtures/database/app/types/mood.sql
@@ -0,0 +1,8 @@
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_type WHERE typname = 'mood' AND typnamespace = (SELECT oid FROM pg_namespace WHERE nspname = 'app')
+    ) THEN
+        CREATE TYPE app.mood AS ENUM ('happy', 'sad');
+    END IF;
+END$$;
diff --git a/tests/testdb/test_schema.py b/tests/testdb/test_schema.py
index 18f3d31..8f71404 100644
--- a/tests/testdb/test_schema.py
+++ b/tests/testdb/test_schema.py
@@ -7,13 +7,34 @@
 
 from pgdevkit.testdb import constants
 from pgdevkit.testdb.container import ensure_container
-from pgdevkit.testdb.schema import apply_schema
+from pgdevkit.testdb.schema import _is_additive_migration, apply_schema
 from tests.testdb.conftest import RUN_SUFFIX, requires_podman
 
 FIXTURES = Path(__file__).parent / "fixtures" / "database"
 TEST_DB = f"pgdevkit_schema_selftest_{RUN_SUFFIX}"
 
 
+@pytest.mark.parametrize(
+    "sql,expected",
+    [
+        ("ALTER TABLE t ADD COLUMN IF NOT EXISTS x text;", True),
+        ("CREATE TABLE IF NOT EXISTS t (id serial primary key);", True),
+        ("CREATE INDEX IF NOT EXISTS idx ON t(x);", True),
+        ("ALTER TABLE t ALTER COLUMN x TYPE text;", False),
+        ("ALTER TABLE t DROP COLUMN x;", False),
+        ("DROP TABLE t;", False),
+        ("DROP TABLE IF EXISTS t;", False),
+        ("TRUNCATE t;", False),
+        ("ALTER TABLE t RENAME COLUMN x TO y;", False),
+        ("ALTER TABLE t RENAME TO t2;", False),
+        ("DELETE FROM t WHERE id = 1;", False),
+        ("ALTER TABLE t OWNER TO someone;", False),
+    ],
+)
+def test_is_additive_migration(sql, expected):
+    assert _is_additive_migration(sql) is expected
+
+
 def _admin_dsn() -> str:
     return constants.conninfo("postgres")
 
@@ -95,3 +116,51 @@ async def test_apply_schema_resolves_view_to_view_dependency(schema_test_db):
             await cur.execute("SELECT id, name FROM app.a_wrapper_view ORDER BY id")
             rows = await cur.fetchall()
     assert rows == [(1, "sprocket")]
+
+
+@requires_podman
+async def test_apply_schema_seeds_composite_enum_and_jsonb_columns(schema_test_db):
+    # gadget.mood is an enum, gadget.size a composite type, gadget.tags
+    # jsonb — plain json.dumps() would corrupt/error on the first two.
+    # gadget.test_data.json also carries a "stale_removed_column" key that
+    # doesn't match any real column (simulating a fixture left behind after
+    # a column rename/removal) — must be silently dropped, not error.
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await apply_schema(con, FIXTURES)
+        async with con.cursor() as cur:
+            await cur.execute("SELECT mood, size, tags FROM app.gadget WHERE id = 1")
+            mood, size, tags = await cur.fetchone()
+    assert mood.name == "happy"
+    assert size == (10, 20)
+    assert tags == ["small", "shiny"]
+
+
+@requires_podman
+async def test_apply_schema_applies_additive_migrations(schema_test_db):
+    # app/migrations/001_add_gadget_note.sql adds a column not present in
+    # gadget.sql itself — only reachable via the migrations pass.
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await apply_schema(con, FIXTURES)
+        async with con.cursor() as cur:
+            await cur.execute(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = 'app' AND table_name = 'gadget' AND column_name = 'note'"
+            )
+            row = await cur.fetchone()
+    assert row is not None
+
+
+@requires_podman
+async def test_apply_schema_skips_unsafe_migrations(schema_test_db):
+    # app/migrations/002_unsafe_drop_gadget_note.sql drops the column that
+    # 001 added — apply_schema must skip it (not execute it), so the column
+    # added by 001 must still be present afterwards.
+    async with await psycopg.AsyncConnection.connect(_db_dsn(), autocommit=True) as con:
+        await apply_schema(con, FIXTURES)
+        async with con.cursor() as cur:
+            await cur.execute(
+                "SELECT column_name FROM information_schema.columns "
+                "WHERE table_schema = 'app' AND table_name = 'gadget' AND column_name = 'note'"
+            )
+            row = await cur.fetchone()
+    assert row is not None
diff --git a/uv.lock b/uv.lock
index 27a8012..d7d03c3 100644
--- a/uv.lock
+++ b/uv.lock
@@ -267,7 +267,7 @@ wheels = [
 
 [[package]]
 name = "pgdevkit"
-version = "0.1.0"
+version = "0.2.0"
 source = { editable = "." }
 dependencies = [
     { name = "psycopg", extra = ["binary"] },