"""Column types backed by PostGIS extensions.
Currently provides :class:`STDADDR`, the ``stdaddr`` composite type
from the ``address_standardizer`` extension. Values round-trip as
:class:`StdAddr` frozen dataclasses -- 16 ``str`` parts where ``""``
stands in for an absent sub-field (stored as NULL).
The ``address_standardizer`` extension must be present in the
database for a ``stdaddr`` column to be created::
CREATE EXTENSION IF NOT EXISTS address_standardizer
"""
from __future__ import annotations
import json
from dataclasses import asdict, dataclass
from typing import Any
from sqlalchemy import cast, func, literal_column
from sqlalchemy import types as sa_types
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.types import UserDefinedType
[docs]
@dataclass(frozen=True, kw_only=True)
class StdAddr:
"""A parsed PostGIS ``stdaddr`` value.
Each attribute maps to one column of the PostgreSQL ``stdaddr``
composite type, in declaration order, and is a ``str`` where
``""`` means the sub-field is absent (stored as NULL). All
fields default to ``""``.
The field set and order mirror the composite type defined by
``address_standardizer``::
CREATE TYPE stdaddr AS (
building text, house_num text, predir text, qual text,
pretype text, name text, suftype text, sufdir text,
ruralroute text, extra text, city text, state text,
country text, postcode text, box text, unit text
);
"""
building: str = ""
house_num: str = ""
predir: str = ""
qual: str = ""
pretype: str = ""
name: str = ""
suftype: str = ""
sufdir: str = ""
ruralroute: str = ""
extra: str = ""
city: str = ""
state: str = ""
country: str = ""
postcode: str = ""
box: str = ""
unit: str = ""
class _StdAddrColumn(UserDefinedType[StdAddr]):
"""The bare ``stdaddr`` SQL type backing :class:`STDADDR`."""
cache_ok = True
def get_col_spec(self, **_kw: Any) -> str: # noqa: ANN401
"""Return the SQL type name."""
return "stdaddr"
[docs]
class STDADDR(sa_types.TypeDecorator[StdAddr]):
"""A PostGIS ``stdaddr`` composite column.
The underlying column is the native ``stdaddr`` type from the
``address_standardizer`` extension. Values round-trip as
:class:`StdAddr` frozen dataclasses; only :class:`StdAddr` (or
``None``) is accepted on input. ``""`` parts are stored as NULL
sub-fields and come back as ``""``.
Conversion is delegated to PostgreSQL via JSON, so there is no
hand-written composite (de)serialization:
- On read, :meth:`column_expression` wraps the column in
``to_jsonb(...)``; the driver returns a ``dict`` that
:meth:`process_result_value` splats into a :class:`StdAddr`.
- On write, :meth:`process_bind_param` emits a JSON object and
:meth:`bind_expression` feeds it through
``jsonb_populate_record(NULL::stdaddr, ...)``.
Requires the extension to be installed::
CREATE EXTENSION IF NOT EXISTS address_standardizer
Example::
from sqlalchemy import Column
from codegen_database.types import STDADDR, StdAddr
Column("address", STDADDR(), nullable=True)
# store StdAddr(house_num="123", name="MAIN", suftype="ST")
"""
impl = _StdAddrColumn
cache_ok = True
[docs]
def column_expression(self, colexpr: Any) -> Any: # noqa: ANN401
"""Read the column as ``jsonb`` so the driver returns a mapping.
``type_=self`` keeps the wrapped expression typed as ``STDADDR``
so :meth:`process_result_value` still runs on the decoded value
(the driver decodes ``jsonb`` to a ``dict`` by its server OID
regardless of the SQLAlchemy type).
"""
return func.to_jsonb(colexpr, type_=self)
[docs]
def process_result_value(
self,
value: Any, # noqa: ANN401
dialect: Any, # noqa: ANN401, ARG002
) -> StdAddr | None:
"""Build a :class:`StdAddr` from the ``to_jsonb`` mapping.
NULL sub-fields become ``""``.
"""
if value is None:
return None
return StdAddr(**{k: (v or "") for k, v in value.items()})
[docs]
def bind_expression(self, bindvalue: Any) -> Any: # noqa: ANN401
"""Build the ``stdaddr`` composite from the bound JSON object."""
return func.jsonb_populate_record(
literal_column("NULL::stdaddr"),
cast(bindvalue, JSONB),
)
[docs]
def process_bind_param(
self,
value: Any, # noqa: ANN401
dialect: Any, # noqa: ANN401, ARG002
) -> str | None:
"""Serialize a :class:`StdAddr` to a JSON object.
``""`` parts become JSON ``null`` (NULL sub-fields). Accepts
any object so a clear error is raised for callers who pass
something other than a :class:`StdAddr`.
"""
if value is None:
return None
if isinstance(value, StdAddr):
return json.dumps(
{k: (v or None) for k, v in asdict(value).items()}
)
msg = f"Expected StdAddr, got {type(value).__name__}"
raise ValueError(msg)