Adopt ExtRef mixin for Stripe identifiers

pkgs/standards/tigrbl/tests/i9n/test_mixins.py

@@ -7,7 +7,7 @@
 import pytest
 from datetime import datetime, timedelta, timezone
 from tigrbl.types import String, uuid4
-from tigrbl.column.shortcuts import acol, F, IO, S
+from tigrbl.column.shortcuts import ColumnSpec, acol, F, IO, S
 
 from tigrbl import Base
 from tigrbl.orm.mixins import (
@@ -147,6 +147,27 @@ class DummyModelExtRef(Base, GUIDPk, ExtRef):
     name = NAME_FIELD
 
 
+class DummyModelExtRefAliased(Base, GUIDPk, ExtRef):
+    """ExtRef mixin with renamed external id/provider columns."""
+
+    __tablename__ = "dummy_ext_ref_aliased"
+    __extref_external_id_attr__ = "stripe_resource_id"
+    __extref_external_id_column__ = "stripe_resource_id"
+    __extref_external_id_spec__ = ColumnSpec(
+        storage=S(type_=String(64), nullable=False, unique=True),
+        field=F(py_type=str, constraints={"max_length": 64}),
+        io=IO(in_verbs=("create",), out_verbs=("read", "list")),
+    )
+    __extref_provider_attr__ = "stripe_provider"
+    __extref_provider_column__ = "stripe_provider"
+    __extref_provider_spec__ = ColumnSpec(
+        storage=S(type_=String(16), default="stripe", nullable=False),
+        field=F(py_type=str, constraints={"max_length": 16}),
+        io=IO(out_verbs=("read", "list")),
+    )
+    name = NAME_FIELD
+
+
 class DummyModelMetaJSON(Base, GUIDPk, MetaJSON):
     """Test model for MetaJSON mixin."""
 
@@ -446,6 +467,32 @@ async def test_ext_ref_mixin(create_test_api):
     assert "external_id" in read_schema.model_fields
 
 
+@pytest.mark.i9n
+@pytest.mark.asyncio
+async def test_ext_ref_mixin_aliases(create_test_api):
+    """ExtRef mixin supports renaming external id/provider columns."""
+
+    create_test_api(DummyModelExtRefAliased)
+
+    create_schema = _build_schema(DummyModelExtRefAliased, verb="create")
+    read_schema = _build_schema(DummyModelExtRefAliased, verb="read")
+
+    assert "stripe_resource_id" in create_schema.model_fields
+    assert "stripe_resource_id" in read_schema.model_fields
+    assert "stripe_provider" in read_schema.model_fields
+    assert "external_id" in read_schema.model_fields
+
+    instance = DummyModelExtRefAliased(name="alias", stripe_resource_id="res_123")
+    assert instance.stripe_resource_id == "res_123"
+    assert instance.external_id == "res_123"
+
+    instance.external_id = "res_456"
+    assert instance.stripe_resource_id == "res_456"
+
+    assert "stripe_resource_id" in DummyModelExtRefAliased.__table__.c
+    assert "stripe_provider" in DummyModelExtRefAliased.__table__.c
+
+
 @pytest.mark.i9n
 @pytest.mark.asyncio
 @pytest.mark.skip(reason="JSONB type not supported in SQLite test environment")

pkgs/standards/tigrbl/tigrbl/orm/mixins/fields.py

@@ -3,6 +3,11 @@
 import datetime as dt
 from decimal import Decimal
 
+from typing import Callable, ClassVar
+
+from sqlalchemy.orm import synonym
+
+from ...column import Column
 from ...specs import ColumnSpec, F, IO, S, acol
 from ...types import (
     TZDateTime,
@@ -100,20 +105,102 @@ class Monetary:
 
 @declarative_mixin
 class ExtRef:
-    external_id: Mapped[str] = acol(
-        spec=ColumnSpec(
-            storage=S(type_=String),
-            field=F(py_type=str),
-            io=CRUD_IO,
-        )
-    )
-    provider: Mapped[str] = acol(
-        spec=ColumnSpec(
-            storage=S(type_=String),
-            field=F(py_type=str),
-            io=CRUD_IO,
+    """Attach an external identifier and provider metadata to a model."""
+
+    #: Attribute name exposed on the SQLAlchemy model for the external id.
+    __extref_external_id_attr__: ClassVar[str] = "external_id"
+    #: Physical column name for the external id – defaults to the attribute name.
+    __extref_external_id_column__: ClassVar[str | None] = None
+    #: Optional :class:`ColumnSpec` override for the external id column.
+    __extref_external_id_spec__: ClassVar[
+        ColumnSpec | Callable[[type], ColumnSpec] | None
+    ] = None
+
+    #: Attribute name used for the provider column.
+    __extref_provider_attr__: ClassVar[str] = "provider"
+    #: Physical column name for the provider – defaults to the attribute name.
+    __extref_provider_column__: ClassVar[str | None] = None
+    #: Optional :class:`ColumnSpec` override for the provider column.
+    __extref_provider_spec__: ClassVar[
+        ColumnSpec | Callable[[type], ColumnSpec] | None
+    ] = None
+
+    @classmethod
+    def __init_subclass__(cls, **kwargs) -> None:
+        super().__init_subclass__(**kwargs)
+        # Only configure concrete subclasses; the mixin itself just holds defaults.
+        if cls is ExtRef:
+            return
+        cls._extref_install()
+
+    @classmethod
+    def _extref_install(cls) -> None:
+        """Ensure external id/provider columns exist with optional aliases."""
+
+        def _resolve(
+            *,
+            spec_config: ColumnSpec | Callable[[type], ColumnSpec] | None,
+            default: ColumnSpec,
+        ) -> ColumnSpec:
+            if callable(spec_config):
+                return spec_config(cls)
+            if spec_config is None:
+                return default
+            return ColumnSpec(
+                storage=spec_config.storage,
+                field=spec_config.field,
+                io=spec_config.io,
+                default_factory=spec_config.default_factory,
+                read_producer=spec_config.read_producer,
+            )
+
+        ext_attr = getattr(cls, "__extref_external_id_attr__", "external_id")
+        ext_col_name = getattr(cls, "__extref_external_id_column__", None) or ext_attr
+        if not isinstance(cls.__dict__.get(ext_attr), Column):
+            ext_spec = _resolve(
+                spec_config=getattr(cls, "__extref_external_id_spec__", None),
+                default=ColumnSpec(
+                    storage=S(type_=String),
+                    field=F(py_type=str),
+                    io=CRUD_IO,
+                ),
+            )
+            ext_col = acol(spec=ext_spec, name=ext_col_name)
+            setattr(cls, ext_attr, ext_col)
+            ext_col.__set_name__(cls, ext_attr)
+        else:
+            ext_col = getattr(cls, ext_attr)
+
+        if ext_attr != "external_id":
+            setattr(cls, "external_id", synonym(ext_attr))
+            colspecs = getattr(cls, "__tigrbl_colspecs__", None)
+            if colspecs is not None and ext_attr in colspecs:
+                colspecs["external_id"] = colspecs[ext_attr]
+
+        provider_attr = getattr(cls, "__extref_provider_attr__", "provider")
+        provider_col_name = (
+            getattr(cls, "__extref_provider_column__", None) or provider_attr
         )
-    )
+        if not isinstance(cls.__dict__.get(provider_attr), Column):
+            provider_spec = _resolve(
+                spec_config=getattr(cls, "__extref_provider_spec__", None),
+                default=ColumnSpec(
+                    storage=S(type_=String),
+                    field=F(py_type=str),
+                    io=CRUD_IO,
+                ),
+            )
+            provider_col = acol(spec=provider_spec, name=provider_col_name)
+            setattr(cls, provider_attr, provider_col)
+            provider_col.__set_name__(cls, provider_attr)
+        else:
+            provider_col = getattr(cls, provider_attr)
+
+        if provider_attr != "provider":
+            setattr(cls, "provider", synonym(provider_attr))
+            colspecs = getattr(cls, "__tigrbl_colspecs__", None)
+            if colspecs is not None and provider_attr in colspecs:
+                colspecs["provider"] = colspecs[provider_attr]
 
 
 @declarative_mixin

pkgs/standards/tigrbl_billing/src/tigrbl_billing/tables/_extref.py

@@ -0,0 +1,52 @@
+"""Helpers for configuring Stripe external identifiers."""
+
+from __future__ import annotations
+
+# Stripe-specific helpers for configuring external identifiers.
+
+from tigrbl.orm.mixins import ExtRef
+from tigrbl.specs import ColumnSpec, F, IO, S
+from tigrbl.types import String
+
+
+class StripeExtRef(ExtRef):
+    """ExtRef mixin preconfigured for Stripe-backed resources."""
+
+    __extref_provider_spec__ = ColumnSpec(
+        storage=S(type_=String(16), nullable=False, default="stripe"),
+        field=F(py_type=str, constraints={"max_length": 16}),
+        io=IO(out_verbs=("read", "list")),
+    )
+
+
+def stripe_external_id_spec(
+    *,
+    length: int = 64,
+    unique: bool = True,
+    index: bool = True,
+    nullable: bool = True,
+    in_verbs: tuple[str, ...] = ("create",),
+    out_verbs: tuple[str, ...] = ("read", "list"),
+    mutable_verbs: tuple[str, ...] | None = None,
+    filter_ops: tuple[str, ...] | None = None,
+) -> ColumnSpec:
+    """Build a :class:`ColumnSpec` for a Stripe external identifier."""
+
+    return ColumnSpec(
+        storage=S(
+            type_=String(length),
+            unique=unique,
+            index=index,
+            nullable=nullable,
+        ),
+        field=F(py_type=str, constraints={"max_length": length}),
+        io=IO(
+            in_verbs=in_verbs,
+            out_verbs=out_verbs,
+            mutable_verbs=mutable_verbs or (),
+            filter_ops=filter_ops or (),
+        ),
+    )
+
+
+__all__ = ["StripeExtRef", "stripe_external_id_spec"]

pkgs/standards/tigrbl_billing/src/tigrbl_billing/tables/application_fee.py

@@ -18,17 +18,20 @@
     UUID,
 )
 
+from ._extref import StripeExtRef, stripe_external_id_spec
 
-class ApplicationFee(Base, GUIDPk, Timestamped):
+
+class ApplicationFee(Base, GUIDPk, Timestamped, StripeExtRef):
     __tablename__ = "application_fees"
 
-    stripe_application_fee_id: Mapped[str | None] = acol(
-        storage=S(type_=String, nullable=True),
-        field=F(py_type=str | None, constraints={"examples": ["fee_123"]}),
-        io=IO(
-            in_verbs=("create", "update", "replace", "merge"),
-            out_verbs=("read", "list"),
-        ),
+    stripe_application_fee_id: Mapped[str | None]
+    __extref_external_id_attr__ = "stripe_application_fee_id"
+    __extref_external_id_column__ = "stripe_application_fee_id"
+    __extref_external_id_spec__ = stripe_external_id_spec(
+        nullable=True,
+        in_verbs=("create", "update", "replace", "merge"),
+        out_verbs=("read", "list"),
+        mutable_verbs=("update", "replace", "merge"),
     )
 
     amount: Mapped[int] = acol(

pkgs/standards/tigrbl_billing/src/tigrbl_billing/tables/balance_top_off.py

@@ -18,6 +18,8 @@
     TZDateTime,
 )
 
+from ._extref import StripeExtRef, stripe_external_id_spec
+
 
 class TopOffTrigger(str, Enum):
     MANUAL = "manual"
@@ -37,7 +39,7 @@ class TopOffStatus(str, Enum):
     FAILED = "failed"
 
 
-class BalanceTopOff(Base, GUIDPk, Timestamped):
+class BalanceTopOff(Base, GUIDPk, Timestamped, StripeExtRef):
     __tablename__ = "balance_top_offs"
 
     balance_id: Mapped[UUID] = acol(
@@ -85,10 +87,16 @@ class BalanceTopOff(Base, GUIDPk, Timestamped):
         io=IO(in_verbs=("create", "update", "replace"), out_verbs=("read", "list")),
     )
 
-    stripe_payment_intent_id: Mapped[str | None] = acol(
-        storage=S(type_=String(128), nullable=True),
-        field=F(py_type=str | None),
-        io=IO(in_verbs=("create", "update", "replace"), out_verbs=("read", "list")),
+    stripe_payment_intent_id: Mapped[str | None]
+    __extref_external_id_attr__ = "stripe_payment_intent_id"
+    __extref_external_id_column__ = "stripe_payment_intent_id"
+    __extref_external_id_spec__ = stripe_external_id_spec(
+        length=128,
+        unique=False,
+        nullable=True,
+        in_verbs=("create", "update", "replace"),
+        out_verbs=("read", "list"),
+        mutable_verbs=("update", "replace"),
     )
 
     processed_at: Mapped[dt.datetime | None] = acol(

pkgs/standards/tigrbl_billing/src/tigrbl_billing/tables/checkout_session.py

@@ -19,6 +19,8 @@
     TZDateTime,
 )
 
+from ._extref import StripeExtRef, stripe_external_id_spec
+
 
 class CheckoutMode(Enum):
     PAYMENT = "payment"
@@ -32,16 +34,17 @@ class CheckoutStatus(Enum):
     EXPIRED = "expired"
 
 
-class CheckoutSession(Base, GUIDPk, Timestamped):
+class CheckoutSession(Base, GUIDPk, Timestamped, StripeExtRef):
     __tablename__ = "checkout_sessions"
 
-    stripe_checkout_session_id: Mapped[str | None] = acol(
-        storage=S(type_=String, nullable=True),
-        field=F(py_type=str | None, constraints={"examples": ["cs_test_123"]}),
-        io=IO(
-            in_verbs=("create", "update", "replace", "merge"),
-            out_verbs=("read", "list"),
-        ),
+    stripe_checkout_session_id: Mapped[str | None]
+    __extref_external_id_attr__ = "stripe_checkout_session_id"
+    __extref_external_id_column__ = "stripe_checkout_session_id"
+    __extref_external_id_spec__ = stripe_external_id_spec(
+        nullable=True,
+        in_verbs=("create", "update", "replace", "merge"),
+        out_verbs=("read", "list"),
+        mutable_verbs=("update", "replace", "merge"),
     )
 
     mode: Mapped[CheckoutMode] = acol(

pkgs/standards/tigrbl_billing/src/tigrbl_billing/tables/connected_account.py

@@ -7,7 +7,9 @@
 from tigrbl.table import Base
 from tigrbl.orm.mixins import GUIDPk, Timestamped
 from tigrbl.specs import F, IO, S, acol
-from tigrbl.types import Mapped, String, JSONB, Boolean, SAEnum, UniqueConstraint
+from tigrbl.types import Mapped, JSONB, Boolean, SAEnum, UniqueConstraint
+
+from ._extref import StripeExtRef, stripe_external_id_spec
 
 
 class ConnectedType(Enum):
@@ -16,16 +18,17 @@ class ConnectedType(Enum):
     CUSTOM = "custom"
 
 
-class ConnectedAccount(Base, GUIDPk, Timestamped):
+class ConnectedAccount(Base, GUIDPk, Timestamped, StripeExtRef):
     __tablename__ = "connected_accounts"
 
-    stripe_account_id: Mapped[str | None] = acol(
-        storage=S(type_=String, nullable=True, unique=True, index=True),
-        field=F(py_type=str | None),
-        io=IO(
-            in_verbs=("create", "update", "replace", "merge"),
-            out_verbs=("read", "list"),
-        ),
+    stripe_account_id: Mapped[str | None]
+    __extref_external_id_attr__ = "stripe_account_id"
+    __extref_external_id_column__ = "stripe_account_id"
+    __extref_external_id_spec__ = stripe_external_id_spec(
+        nullable=True,
+        in_verbs=("create", "update", "replace", "merge"),
+        out_verbs=("read", "list"),
+        mutable_verbs=("update", "replace", "merge"),
     )
 
     type: Mapped[ConnectedType] = acol(