Initialize identity module

src/govoplan_identity/__init__.py

@@ -0,0 +1,3 @@
+"""GovOPlaN identity module."""
+
+__version__ = "0.1.6"

src/govoplan_identity/backend/__init__.py

@@ -0,0 +1 @@
+"""Backend integration for the GovOPlaN identity module."""

src/govoplan_identity/backend/db/__init__.py

@@ -0,0 +1 @@
+"""Identity database models."""

src/govoplan_identity/backend/db/models.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import Boolean, ForeignKey, Index, JSON, String, UniqueConstraint, text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from govoplan_core.db.base import Base, TimestampMixin
+
+
+def new_uuid() -> str:
+    return str(uuid.uuid4())
+
+
+class CanonicalIdentity(Base, TimestampMixin):
+    __tablename__ = "identity_identities"
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    display_name: Mapped[str | None] = mapped_column(String(255))
+    external_subject: Mapped[str | None] = mapped_column(String(255), index=True)
+    source: Mapped[str] = mapped_column(String(50), default="local", nullable=False)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
+    settings: Mapped[dict[str, Any]] = mapped_column(JSON, default=dict, nullable=False)
+
+
+class CanonicalIdentityAccountLink(Base, TimestampMixin):
+    __tablename__ = "identity_account_links"
+    __table_args__ = (
+        UniqueConstraint("identity_id", "account_id", name="uq_identity_module_account_links_identity_account"),
+        Index(
+            "uq_identity_module_account_links_primary_account",
+            "account_id",
+            unique=True,
+            sqlite_where=text("is_primary = 1"),
+            postgresql_where=text("is_primary IS TRUE"),
+        ),
+        Index(
+            "uq_identity_module_account_links_primary_identity",
+            "identity_id",
+            unique=True,
+            sqlite_where=text("is_primary = 1"),
+            postgresql_where=text("is_primary IS TRUE"),
+        ),
+    )
+
+    id: Mapped[str] = mapped_column(String(36), primary_key=True, default=new_uuid)
+    identity_id: Mapped[str] = mapped_column(ForeignKey("identity_identities.id", ondelete="CASCADE"), nullable=False, index=True)
+    account_id: Mapped[str] = mapped_column(String(36), nullable=False, index=True)
+    is_primary: Mapped[bool] = mapped_column(Boolean, default=False, nullable=False)
+    source: Mapped[str] = mapped_column(String(50), default="local", nullable=False)
+
+
+Identity = CanonicalIdentity
+IdentityAccountLink = CanonicalIdentityAccountLink
+
+
+__all__ = ["CanonicalIdentity", "CanonicalIdentityAccountLink", "Identity", "IdentityAccountLink", "new_uuid"]

src/govoplan_identity/backend/directory.py

@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityDirectory, IdentityRef
+from govoplan_core.db.session import get_database
+from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
+
+
+def _status(active: bool) -> str:
+    return "active" if active else "inactive"
+
+
+def _identity_ref(identity: Identity, links: Sequence[IdentityAccountLink]) -> IdentityRef:
+    primary_account_id = next((link.account_id for link in links if link.is_primary), None)
+    return IdentityRef(
+        id=identity.id,
+        display_name=identity.display_name,
+        external_subject=identity.external_subject,
+        source=identity.source,
+        primary_account_id=primary_account_id,
+        account_ids=tuple(link.account_id for link in links),
+        status=_status(identity.is_active),  # type: ignore[arg-type]
+    )
+
+
+def _link_ref(link: IdentityAccountLink) -> IdentityAccountLinkRef:
+    return IdentityAccountLinkRef(
+        id=link.id,
+        identity_id=link.identity_id,
+        account_id=link.account_id,
+        is_primary=link.is_primary,
+        source=link.source,
+    )
+
+
+class SqlIdentityDirectory(IdentityDirectory):
+    def get_identity(self, identity_id: str) -> IdentityRef | None:
+        with get_database().session() as session:
+            identity = session.get(Identity, identity_id)
+            if identity is None:
+                return None
+            links = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.identity_id == identity.id)
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.created_at.asc())
+                .all()
+            )
+            return _identity_ref(identity, links)
+
+    def identity_for_account(self, account_id: str) -> IdentityRef | None:
+        with get_database().session() as session:
+            link = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.account_id == account_id)
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.created_at.asc())
+                .first()
+            )
+            if link is None:
+                return None
+            identity = session.get(Identity, link.identity_id)
+            if identity is None:
+                return None
+            links = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.identity_id == identity.id)
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.created_at.asc())
+                .all()
+            )
+            return _identity_ref(identity, links)
+
+    def identities_for_accounts(self, account_ids: Sequence[str]) -> tuple[IdentityRef, ...]:
+        ids = sorted({str(account_id) for account_id in account_ids if account_id})
+        if not ids:
+            return ()
+        with get_database().session() as session:
+            identity_ids = [
+                row[0]
+                for row in session.query(IdentityAccountLink.identity_id)
+                .filter(IdentityAccountLink.account_id.in_(ids))
+                .distinct()
+                .all()
+            ]
+            identities = session.query(Identity).filter(Identity.id.in_(identity_ids)).all() if identity_ids else []
+            links = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.identity_id.in_(identity_ids))
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.created_at.asc())
+                .all()
+                if identity_ids else []
+            )
+            links_by_identity: dict[str, list[IdentityAccountLink]] = {}
+            for link in links:
+                links_by_identity.setdefault(link.identity_id, []).append(link)
+            return tuple(_identity_ref(identity, links_by_identity.get(identity.id, [])) for identity in identities)
+
+    def accounts_for_identity(self, identity_id: str) -> tuple[IdentityAccountLinkRef, ...]:
+        with get_database().session() as session:
+            links = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.identity_id == identity_id)
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.created_at.asc())
+                .all()
+            )
+            return tuple(_link_ref(link) for link in links)

src/govoplan_identity/backend/manifest.py

@@ -0,0 +1,51 @@
+from __future__ import annotations
+
+from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY
+from govoplan_core.core.module_guards import persistent_table_uninstall_guard
+from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest
+from govoplan_core.db.base import Base
+from govoplan_identity.backend.db import models as identity_models  # noqa: F401 - populate metadata
+
+
+def _identity_directory(context: ModuleContext) -> object:
+    del context
+    from govoplan_identity.backend.directory import SqlIdentityDirectory
+
+    return SqlIdentityDirectory()
+
+
+manifest = ModuleManifest(
+    id="identity",
+    name="Identity",
+    version="0.1.6",
+    migration_spec=MigrationSpec(module_id="identity", metadata=Base.metadata),
+    uninstall_guard_providers=(
+        persistent_table_uninstall_guard(
+            identity_models.Identity,
+            identity_models.IdentityAccountLink,
+            label="Identity",
+        ),
+    ),
+    capability_factories={
+        CAPABILITY_IDENTITY_DIRECTORY: _identity_directory,
+    },
+    documentation=(
+        DocumentationTopic(
+            id="identity.model",
+            title="Identity directory",
+            summary="Identity owns normalized subjects and links them to platform accounts. Access owns authorization.",
+            body=(
+                "An identity can have multiple accounts. Identity facts remain separate from authentication sessions, "
+                "organization functions, and permission decisions."
+            ),
+            layer="configured",
+            documentation_types=("admin", "user"),
+            audience=("tenant_admin", "access_admin", "operator"),
+            order=24,
+        ),
+    ),
+)
+
+
+def get_manifest() -> ModuleManifest:
+    return manifest

src/govoplan_identity/py.typed

@@ -0,0 +1 @@
+