diff --git a/src/music_catalogue/__init__.py b/src/music_catalogue/__init__.py index 7cd229a..8bbf0f7 100644 --- a/src/music_catalogue/__init__.py +++ b/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, ) diff --git a/src/music_catalogue/_generated/_register.py b/src/music_catalogue/_generated/_register.py deleted file mode 100644 index a4e6dc1..0000000 --- a/src/music_catalogue/_generated/_register.py +++ /dev/null @@ -1,139 +0,0 @@ -# @generated by python.gen (pGenie); regeneration overwrites manual changes. -# SPDX-FileCopyrightText: 2026 Viacheslav Shvets -# SPDX-License-Identifier: MIT-0 - -from __future__ import annotations - -import keyword -from collections.abc import Callable, Sequence -from dataclasses import fields, is_dataclass -from typing import Any - -from psycopg import AsyncConnection, Connection -from psycopg.types.composite import CompositeInfo, register_composite -from psycopg.types.enum import EnumInfo, register_enum - -from . import types as _db_types - -type _ObjectMaker[T] = Callable[[Sequence[Any], CompositeInfo], T] -type _SequenceMaker[T] = Callable[[T, CompositeInfo], Sequence[Any]] - - -def _python_name(name: str) -> str: - if keyword.iskeyword(name): - return f"{name}_" - return name - - -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 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 - - -_album_format_pg_name = "public.album_format" -_recording_info_pg_name = "public.recording_info" -_recording_info_make_object, _recording_info_make_sequence = _dataclass_callbacks(_db_types.RecordingInfo) -_track_info_pg_name = "public.track_info" -_track_info_make_object, _track_info_make_sequence = _dataclass_callbacks(_db_types.TrackInfo) -_disc_info_pg_name = "public.disc_info" -_disc_info_make_object, _disc_info_make_sequence = _dataclass_callbacks(_db_types.DiscInfo) - - -async def register_types(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, - _db_types.AlbumFormat, - mapping={member: member.value for member in _db_types.AlbumFormat}, - ) - 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, - _db_types.RecordingInfo, - make_object=_recording_info_make_object, - make_sequence=_recording_info_make_sequence, - ) - track_info_info = await CompositeInfo.fetch(conn, _track_info_pg_name) - if track_info_info is None: - raise LookupError(f"composite type {_track_info_pg_name!r} not found; cannot register it") - register_composite( - track_info_info, - conn, - _db_types.TrackInfo, - make_object=_track_info_make_object, - make_sequence=_track_info_make_sequence, - ) - 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, - _db_types.DiscInfo, - make_object=_disc_info_make_object, - make_sequence=_disc_info_make_sequence, - ) - - -def register_types_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, - _db_types.AlbumFormat, - mapping={member: member.value for member in _db_types.AlbumFormat}, - ) - 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, - _db_types.RecordingInfo, - make_object=_recording_info_make_object, - make_sequence=_recording_info_make_sequence, - ) - track_info_info = CompositeInfo.fetch(conn, _track_info_pg_name) - if track_info_info is None: - raise LookupError(f"composite type {_track_info_pg_name!r} not found; cannot register it") - register_composite( - track_info_info, - conn, - _db_types.TrackInfo, - make_object=_track_info_make_object, - make_sequence=_track_info_make_sequence, - ) - 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, - _db_types.DiscInfo, - make_object=_disc_info_make_object, - make_sequence=_disc_info_make_sequence, - ) diff --git a/src/music_catalogue/_generated/_runtime.py b/src/music_catalogue/_generated/_runtime.py index 625870a..ceadcc3 100644 --- a/src/music_catalogue/_generated/_runtime.py +++ b/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 diff --git a/src/music_catalogue/_generated/types/__init__.py b/src/music_catalogue/_generated/types/__init__.py index 20069be..d0d0ddc 100644 --- a/src/music_catalogue/_generated/types/__init__.py +++ b/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) diff --git a/src/music_catalogue/_generated/types/album_format.py b/src/music_catalogue/_generated/types/album_format.py index d6165bd..d58d211 100644 --- a/src/music_catalogue/_generated/types/album_format.py +++ b/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}, + ) diff --git a/src/music_catalogue/_generated/types/disc_info.py b/src/music_catalogue/_generated/types/disc_info.py index bcdab8b..1f9e1e5 100644 --- a/src/music_catalogue/_generated/types/disc_info.py +++ b/src/music_catalogue/_generated/types/disc_info.py @@ -4,6 +4,10 @@ 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 @@ -11,3 +15,33 @@ 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, + ) diff --git a/src/music_catalogue/_generated/types/recording_info.py b/src/music_catalogue/_generated/types/recording_info.py index 94c7afa..3b6c848 100644 --- a/src/music_catalogue/_generated/types/recording_info.py +++ b/src/music_catalogue/_generated/types/recording_info.py @@ -6,6 +6,11 @@ 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: @@ -13,3 +18,33 @@ class RecordingInfo: 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, + ) diff --git a/src/music_catalogue/_generated/types/track_info.py b/src/music_catalogue/_generated/types/track_info.py index 88353d4..51f1df2 100644 --- a/src/music_catalogue/_generated/types/track_info.py +++ b/src/music_catalogue/_generated/types/track_info.py @@ -4,9 +4,44 @@ from dataclasses import dataclass +from psycopg import AsyncConnection, Connection +from psycopg.types.composite import CompositeInfo, register_composite + +from .. import _runtime + @dataclass(frozen=True, slots=True) class TrackInfo: title: str duration_seconds: int tags: list[str] + + +_track_info_pg_name = "public.track_info" +_track_info_make_object, _track_info_make_sequence = _runtime.dataclass_callbacks(TrackInfo) + + +async def register(conn: AsyncConnection[object]) -> None: + track_info_info = await CompositeInfo.fetch(conn, _track_info_pg_name) + if track_info_info is None: + raise LookupError(f"composite type {_track_info_pg_name!r} not found; cannot register it") + register_composite( + track_info_info, + conn, + TrackInfo, + make_object=_track_info_make_object, + make_sequence=_track_info_make_sequence, + ) + + +def register_sync(conn: Connection[object]) -> None: + track_info_info = CompositeInfo.fetch(conn, _track_info_pg_name) + if track_info_info is None: + raise LookupError(f"composite type {_track_info_pg_name!r} not found; cannot register it") + register_composite( + track_info_info, + conn, + TrackInfo, + make_object=_track_info_make_object, + make_sequence=_track_info_make_sequence, + ) diff --git a/src/music_catalogue/sync/__init__.py b/src/music_catalogue/sync/__init__.py index 237544e..aceaa02 100644 --- a/src/music_catalogue/sync/__init__.py +++ b/src/music_catalogue/sync/__init__.py @@ -8,9 +8,6 @@ from .._generated._core import ( NoRowError as NoRowError, ) -from .._generated._register import ( - register_types_sync as register_types, -) from .._generated.statements.insert_album import ( InsertAlbumRow as InsertAlbumRow, ) @@ -68,6 +65,9 @@ from .._generated.statements.update_album_released import ( update_album_released_sync as update_album_released, ) +from .._generated.types import ( + register_types_sync as register_types, +) from .._generated.types.album_format import ( AlbumFormat as AlbumFormat, )