feat(identity): expose directory search capability

src/govoplan_identity/backend/directory.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 from collections.abc import Sequence
 
+from sqlalchemy import func, or_
+
 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
@@ -103,3 +105,49 @@ class SqlIdentityDirectory(IdentityDirectory):
                 .all()
             )
             return tuple(_link_ref(link) for link in links)
+
+    def search_identities(
+        self,
+        query: str | None = None,
+        *,
+        include_inactive: bool = False,
+        limit: int = 25,
+    ) -> tuple[IdentityRef, ...]:
+        normalized_limit = max(1, min(int(limit), 100))
+        with get_database().session() as session:
+            identity_query = session.query(Identity)
+            if not include_inactive:
+                identity_query = identity_query.filter(Identity.is_active.is_(True))
+            if query and query.strip():
+                pattern = f"%{query.strip().casefold()}%"
+                matching_account_links = session.query(IdentityAccountLink.identity_id).filter(
+                    func.lower(IdentityAccountLink.account_id).like(pattern)
+                )
+                identity_query = identity_query.filter(
+                    or_(
+                        func.lower(Identity.id).like(pattern),
+                        func.lower(Identity.display_name).like(pattern),
+                        func.lower(Identity.external_subject).like(pattern),
+                        Identity.id.in_(matching_account_links),
+                    )
+                )
+
+            identities = (
+                identity_query
+                .order_by(Identity.display_name.asc(), Identity.id.asc())
+                .limit(normalized_limit)
+                .all()
+            )
+            identity_ids = [identity.id for identity in identities]
+            links = (
+                session.query(IdentityAccountLink)
+                .filter(IdentityAccountLink.identity_id.in_(identity_ids))
+                .order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.account_id.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)

src/govoplan_identity/backend/manifest.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from pathlib import Path
 
 from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER
-from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY
+from govoplan_core.core.identity import CAPABILITY_IDENTITY_DIRECTORY, CAPABILITY_IDENTITY_SEARCH
 from govoplan_core.core.module_guards import persistent_table_uninstall_guard
 from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest, PermissionDefinition, RoleTemplate
 from govoplan_core.db.base import Base
@@ -75,6 +75,7 @@ manifest = ModuleManifest(
     ),
     capability_factories={
         CAPABILITY_IDENTITY_DIRECTORY: _identity_directory,
+        CAPABILITY_IDENTITY_SEARCH: _identity_directory,
     },
     documentation=(
         DocumentationTopic(

tests/test_directory.py

@@ -0,0 +1,83 @@
+from __future__ import annotations
+
+import unittest
+
+from govoplan_core.core.identity import IdentitySearchProvider
+from govoplan_core.db.base import Base
+from govoplan_core.db.session import configure_database, reset_database
+from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
+from govoplan_identity.backend.directory import SqlIdentityDirectory
+
+
+class IdentityDirectoryTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.database = configure_database("sqlite:///:memory:")
+        Base.metadata.create_all(
+            self.database.engine,
+            tables=[Identity.__table__, IdentityAccountLink.__table__],
+        )
+        with self.database.session() as session:
+            active = Identity(
+                id="identity-active",
+                display_name="Ada Example",
+                external_subject="subject-active",
+                source="local",
+                is_active=True,
+                settings={},
+            )
+            inactive = Identity(
+                id="identity-inactive",
+                display_name="Inactive Person",
+                external_subject="subject-inactive",
+                source="local",
+                is_active=False,
+                settings={},
+            )
+            session.add_all([active, inactive])
+            session.flush()
+            session.add_all(
+                [
+                    IdentityAccountLink(
+                        id="link-active",
+                        identity_id=active.id,
+                        account_id="account-ada",
+                        is_primary=True,
+                        source="local",
+                    ),
+                    IdentityAccountLink(
+                        id="link-inactive",
+                        identity_id=inactive.id,
+                        account_id="account-inactive",
+                        is_primary=True,
+                        source="local",
+                    ),
+                ]
+            )
+            session.commit()
+
+    def tearDown(self) -> None:
+        reset_database(dispose=True)
+
+    def test_search_capability_filters_inactive_identities_and_searches_account_ids(self) -> None:
+        directory = SqlIdentityDirectory()
+
+        self.assertIsInstance(directory, IdentitySearchProvider)
+        self.assertEqual(
+            ("identity-active",),
+            tuple(identity.id for identity in directory.search_identities()),
+        )
+        matches = directory.search_identities("account-ada")
+        self.assertEqual(("identity-active",), tuple(identity.id for identity in matches))
+        self.assertEqual(("account-ada",), matches[0].account_ids)
+        self.assertEqual("account-ada", matches[0].primary_account_id)
+
+    def test_search_capability_can_include_inactive_identities(self) -> None:
+        directory = SqlIdentityDirectory()
+
+        matches = directory.search_identities("inactive", include_inactive=True)
+
+        self.assertEqual(("identity-inactive",), tuple(identity.id for identity in matches))
+
+
+if __name__ == "__main__":
+    unittest.main()