106 lines
4.4 KiB
Python
106 lines
4.4 KiB
Python
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)
|