3 Commits
v0.1.7 ... main

9 changed files with 194 additions and 5 deletions

View File

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

View File

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

View File

@@ -1,3 +1,3 @@
"""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 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)

View File

@@ -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
@@ -55,7 +55,7 @@ def _identity_directory(context: ModuleContext) -> object:
manifest = ModuleManifest(
id="identity",
name="Identity",
version="0.1.7",
version="0.1.8",
required_capabilities=(CAPABILITY_AUTH_PRINCIPAL_RESOLVER, CAPABILITY_AUTH_PERMISSION_EVALUATOR),
permissions=PERMISSIONS,
role_templates=ROLE_TEMPLATES,
@@ -75,6 +75,7 @@ manifest = ModuleManifest(
),
capability_factories={
CAPABILITY_IDENTITY_DIRECTORY: _identity_directory,
CAPABILITY_IDENTITY_SEARCH: _identity_directory,
},
documentation=(
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()