Move registration to where the types are declared

src/music_catalogue/__init__.py

@@ -11,9 +11,6 @@
 from ._generated._core import (
     NoRowError as NoRowError,
 )
-from ._generated._register import (
-    register_types as register_types,
-)
 from ._generated.statements.insert_album import (
     InsertAlbumRow as InsertAlbumRow,
 )
@@ -71,6 +68,9 @@
 from ._generated.statements.update_album_released import (
     update_album_released as update_album_released,
 )
+from ._generated.types import (
+    register_types as register_types,
+)
 from ._generated.types.album_format import (
     AlbumFormat as AlbumFormat,
 )

src/music_catalogue/_generated/_runtime.py

@@ -4,10 +4,14 @@
 
 from __future__ import annotations
 
-from typing import LiteralString
+import keyword
+from collections.abc import Callable, Sequence
+from dataclasses import fields, is_dataclass
+from typing import Any, LiteralString
 
 from psycopg import AsyncConnection
 from psycopg.rows import BaseRowFactory
+from psycopg.types.composite import CompositeInfo
 
 from ._core import NoRowError
 
@@ -67,3 +71,33 @@ async def execute_void(
 ) -> None:
     async with conn.cursor() as cur:
         _ = await cur.execute(sql, params)
+
+
+type _ObjectMaker[T] = Callable[[Sequence[Any], CompositeInfo], T]
+type _SequenceMaker[T] = Callable[[T, CompositeInfo], Sequence[Any]]
+
+
+def dataclass_callbacks[T](cls: type[T]) -> tuple[_ObjectMaker[T], _SequenceMaker[T]]:
+    if not is_dataclass(cls):
+        raise TypeError(f"{cls.__name__} must be a dataclass")
+
+    model_fields = fields(cls)
+    model_names = tuple(field.name for field in model_fields)
+
+    def _python_name(name: str) -> str:
+        if keyword.iskeyword(name):
+            return f"{name}_"
+        return name
+
+    def make_object(values: Sequence[Any], info: CompositeInfo) -> T:
+        names = tuple(_python_name(name) for name in info.field_names)
+        assert names == model_names
+        assert len(values) == len(model_fields)
+        return cls(**dict(zip(names, values, strict=True)))
+
+    def make_sequence(obj: T, info: CompositeInfo) -> Sequence[Any]:
+        names = tuple(_python_name(name) for name in info.field_names)
+        assert names == model_names
+        return tuple(getattr(obj, field.name) for field in model_fields)
+
+    return make_object, make_sequence

src/music_catalogue/_generated/types/__init__.py

@@ -2,7 +2,25 @@
 # SPDX-FileCopyrightText: 2026 Viacheslav Shvets
 # SPDX-License-Identifier: MIT-0
 
+from __future__ import annotations
+
+from psycopg import AsyncConnection, Connection
+
 from .album_format import AlbumFormat as AlbumFormat
 from .disc_info import DiscInfo as DiscInfo
 from .recording_info import RecordingInfo as RecordingInfo
 from .track_info import TrackInfo as TrackInfo
+
+
+async def register_types(conn: AsyncConnection[object]) -> None:
+    await album_format.register(conn)
+    await recording_info.register(conn)
+    await track_info.register(conn)
+    await disc_info.register(conn)
+
+
+def register_types_sync(conn: Connection[object]) -> None:
+    album_format.register_sync(conn)
+    recording_info.register_sync(conn)
+    track_info.register_sync(conn)
+    disc_info.register_sync(conn)

src/music_catalogue/_generated/types/album_format.py

@@ -4,6 +4,9 @@
 
 from enum import StrEnum
 
+from psycopg import AsyncConnection, Connection
+from psycopg.types.enum import EnumInfo, register_enum
+
 
 class AlbumFormat(StrEnum):
     VINYL = "Vinyl"
@@ -12,3 +15,30 @@ class AlbumFormat(StrEnum):
     DIGITAL = "Digital"
     DVD_AUDIO = "DVD-Audio"
     SACD = "SACD"
+
+
+_album_format_pg_name = "public.album_format"
+
+
+async def register(conn: AsyncConnection[object]) -> None:
+    album_format_info = await EnumInfo.fetch(conn, _album_format_pg_name)
+    if album_format_info is None:
+        raise LookupError(f"enum type {_album_format_pg_name!r} not found; cannot register it")
+    register_enum(
+        album_format_info,
+        conn,
+        AlbumFormat,
+        mapping={member: member.value for member in AlbumFormat},
+    )
+
+
+def register_sync(conn: Connection[object]) -> None:
+    album_format_info = EnumInfo.fetch(conn, _album_format_pg_name)
+    if album_format_info is None:
+        raise LookupError(f"enum type {_album_format_pg_name!r} not found; cannot register it")
+    register_enum(
+        album_format_info,
+        conn,
+        AlbumFormat,
+        mapping={member: member.value for member in AlbumFormat},
+    )

src/music_catalogue/_generated/types/disc_info.py

@@ -4,10 +4,44 @@
 
 from dataclasses import dataclass
 
+from psycopg import AsyncConnection, Connection
+from psycopg.types.composite import CompositeInfo, register_composite
+
+from .. import _runtime
 from .recording_info import RecordingInfo
 
 
 @dataclass(frozen=True, slots=True)
 class DiscInfo:
     name: str | None
     recording: RecordingInfo | None
+
+
+_disc_info_pg_name = "public.disc_info"
+_disc_info_make_object, _disc_info_make_sequence = _runtime.dataclass_callbacks(DiscInfo)
+
+
+async def register(conn: AsyncConnection[object]) -> None:
+    disc_info_info = await CompositeInfo.fetch(conn, _disc_info_pg_name)
+    if disc_info_info is None:
+        raise LookupError(f"composite type {_disc_info_pg_name!r} not found; cannot register it")
+    register_composite(
+        disc_info_info,
+        conn,
+        DiscInfo,
+        make_object=_disc_info_make_object,
+        make_sequence=_disc_info_make_sequence,
+    )
+
+
+def register_sync(conn: Connection[object]) -> None:
+    disc_info_info = CompositeInfo.fetch(conn, _disc_info_pg_name)
+    if disc_info_info is None:
+        raise LookupError(f"composite type {_disc_info_pg_name!r} not found; cannot register it")
+    register_composite(
+        disc_info_info,
+        conn,
+        DiscInfo,
+        make_object=_disc_info_make_object,
+        make_sequence=_disc_info_make_sequence,
+    )

src/music_catalogue/_generated/types/recording_info.py

@@ -6,10 +6,45 @@
 
 from datetime import date
 
+from psycopg import AsyncConnection, Connection
+from psycopg.types.composite import CompositeInfo, register_composite
+
+from .. import _runtime
+
 
 @dataclass(frozen=True, slots=True)
 class RecordingInfo:
     studio_name: str
     city: str
     country: str
     recorded_date: date
+
+
+_recording_info_pg_name = "public.recording_info"
+_recording_info_make_object, _recording_info_make_sequence = _runtime.dataclass_callbacks(RecordingInfo)
+
+
+async def register(conn: AsyncConnection[object]) -> None:
+    recording_info_info = await CompositeInfo.fetch(conn, _recording_info_pg_name)
+    if recording_info_info is None:
+        raise LookupError(f"composite type {_recording_info_pg_name!r} not found; cannot register it")
+    register_composite(
+        recording_info_info,
+        conn,
+        RecordingInfo,
+        make_object=_recording_info_make_object,
+        make_sequence=_recording_info_make_sequence,
+    )
+
+
+def register_sync(conn: Connection[object]) -> None:
+    recording_info_info = CompositeInfo.fetch(conn, _recording_info_pg_name)
+    if recording_info_info is None:
+        raise LookupError(f"composite type {_recording_info_pg_name!r} not found; cannot register it")
+    register_composite(
+        recording_info_info,
+        conn,
+        RecordingInfo,
+        make_object=_recording_info_make_object,
+        make_sequence=_recording_info_make_sequence,
+    )