3 Commits
v0.1.7 ... main

9 changed files with 194 additions and 5 deletions

View File

@@ -1,5 +1,9 @@
# GovOPlaN Identity # GovOPlaN Identity
<!-- govoplan-repository-type:start -->
**Repository type:** module (platform).
<!-- govoplan-repository-type:end -->
`govoplan-identity` is the canonical identity directory module for GovOPlaN. `govoplan-identity` is the canonical identity directory module for GovOPlaN.
It owns identities and links between identities and platform accounts. This is It owns identities and links between identities and platform accounts. This is

View File

@@ -4,13 +4,13 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "govoplan-identity" name = "govoplan-identity"
version = "0.1.7" version = "0.1.8"
description = "GovOPlaN identity directory module." description = "GovOPlaN identity directory module."
readme = "README.md" readme = "README.md"
requires-python = ">=3.12" requires-python = ">=3.12"
authors = [{ name = "GovOPlaN" }] authors = [{ name = "GovOPlaN" }]
dependencies = [ dependencies = [
"govoplan-core>=0.1.7", "govoplan-core>=0.1.8",
] ]
[tool.setuptools.packages.find] [tool.setuptools.packages.find]

View File

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

View File

@@ -2,6 +2,8 @@ from __future__ import annotations
from collections.abc import Sequence from collections.abc import Sequence
from sqlalchemy import func, or_
from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityDirectory, IdentityRef from govoplan_core.core.identity import IdentityAccountLinkRef, IdentityDirectory, IdentityRef
from govoplan_core.db.session import get_database from govoplan_core.db.session import get_database
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
@@ -103,3 +105,49 @@ class SqlIdentityDirectory(IdentityDirectory):
.all() .all()
) )
return tuple(_link_ref(link) for link in links) 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)

View File

@@ -3,7 +3,7 @@ from __future__ import annotations
from pathlib import Path from pathlib import Path
from govoplan_core.core.access import CAPABILITY_AUTH_PERMISSION_EVALUATOR, CAPABILITY_AUTH_PRINCIPAL_RESOLVER 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.module_guards import persistent_table_uninstall_guard
from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest, PermissionDefinition, RoleTemplate from govoplan_core.core.modules import DocumentationTopic, MigrationSpec, ModuleContext, ModuleManifest, PermissionDefinition, RoleTemplate
from govoplan_core.db.base import Base from govoplan_core.db.base import Base
@@ -55,7 +55,7 @@ def _identity_directory(context: ModuleContext) -> object:
manifest = ModuleManifest( manifest = ModuleManifest(
id="identity", id="identity",
name="Identity", name="Identity",
version="0.1.7", version="0.1.8",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR), required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
permissions=PERMISSIONS, permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES, role_templates=ROLE_TEMPLATES,
@@ -75,6 +75,7 @@ manifest = ModuleManifest(
), ),
capability_factories={ capability_factories={
CAPABILITY_IDENTITY_DIRECTORY: _identity_directory, CAPABILITY_IDENTITY_DIRECTORY: _identity_directory,
CAPABILITY_IDENTITY_SEARCH: _identity_directory,
}, },
documentation=( documentation=(
DocumentationTopic( DocumentationTopic(

View File

@@ -0,0 +1 @@
"""Identity module migration versions."""

View File

@@ -0,0 +1,52 @@
"""v0.1.7 identity baseline
Revision ID: 5c6d7e8f9a10
Revises: None
Create Date: 2026-07-11 00:00:00.000000
"""
from __future__ import annotations
from alembic import op
import sqlalchemy as sa
revision = '5c6d7e8f9a10'
down_revision = None
branch_labels = None
depends_on = ('4f2a9c8e7b6d', '4a5b6c7d8e9f')
def upgrade() -> None:
op.create_table('identity_identities',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('display_name', sa.String(length=255), nullable=True),
sa.Column('external_subject', sa.String(length=255), nullable=True),
sa.Column('source', sa.String(length=50), nullable=False),
sa.Column('is_active', sa.Boolean(), nullable=False),
sa.Column('settings', sa.JSON(), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
sa.PrimaryKeyConstraint('id', name=op.f('pk_identity_identities'))
)
op.create_index(op.f('ix_identity_identities_external_subject'), 'identity_identities', ['external_subject'], unique=False)
op.create_table('identity_account_links',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('identity_id', sa.String(length=36), nullable=False),
sa.Column('account_id', sa.String(length=36), nullable=False),
sa.Column('is_primary', sa.Boolean(), nullable=False),
sa.Column('source', sa.String(length=50), nullable=False),
sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
sa.ForeignKeyConstraint(['identity_id'], ['identity_identities.id'], name=op.f('fk_identity_account_links_identity_id_identity_identities'), ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id', name=op.f('pk_identity_account_links')),
sa.UniqueConstraint('identity_id', 'account_id', name='uq_identity_module_account_links_identity_account')
)
op.create_index(op.f('ix_identity_account_links_account_id'), 'identity_account_links', ['account_id'], unique=False)
op.create_index(op.f('ix_identity_account_links_identity_id'), 'identity_account_links', ['identity_id'], unique=False)
op.create_index('uq_identity_module_account_links_primary_account', 'identity_account_links', ['account_id'], unique=True, sqlite_where=sa.text('is_primary = 1'), postgresql_where=sa.text('is_primary IS TRUE'))
op.create_index('uq_identity_module_account_links_primary_identity', 'identity_account_links', ['identity_id'], unique=True, sqlite_where=sa.text('is_primary = 1'), postgresql_where=sa.text('is_primary IS TRUE'))
def downgrade() -> None:
op.drop_table('identity_account_links')
op.drop_table('identity_identities')

83
tests/test_directory.py Normal file
View File

@@ -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()