83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal, require_any_scope
|
|
from govoplan_core.db.session import get_session
|
|
from govoplan_identity.backend.db.models import Identity, IdentityAccountLink
|
|
|
|
from .schemas import IdentityItem, IdentityListResponse
|
|
|
|
|
|
router = APIRouter(prefix="/identity", tags=["identity"])
|
|
|
|
IDENTITY_READ_SCOPES = (
|
|
"identity:identity:read",
|
|
"identity:read",
|
|
"admin:users:read",
|
|
"system:accounts:read",
|
|
"organizations:function:assign",
|
|
)
|
|
|
|
|
|
@router.get("/identities", response_model=IdentityListResponse)
|
|
def list_identities(
|
|
query: str | None = Query(default=None, min_length=1, max_length=255),
|
|
limit: int = Query(default=25, ge=1, le=100),
|
|
include_inactive: bool = False,
|
|
session: Session = Depends(get_session),
|
|
principal: ApiPrincipal = Depends(require_any_scope(*IDENTITY_READ_SCOPES)),
|
|
) -> IdentityListResponse:
|
|
del principal
|
|
identity_query = session.query(Identity)
|
|
if not include_inactive:
|
|
identity_query = identity_query.filter(Identity.is_active.is_(True))
|
|
if query:
|
|
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(limit).all()
|
|
identity_ids = [identity.id for identity in identities]
|
|
links_by_identity: dict[str, list[IdentityAccountLink]] = {identity_id: [] for identity_id in identity_ids}
|
|
if identity_ids:
|
|
links = (
|
|
session.query(IdentityAccountLink)
|
|
.filter(IdentityAccountLink.identity_id.in_(identity_ids))
|
|
.order_by(IdentityAccountLink.is_primary.desc(), IdentityAccountLink.account_id.asc())
|
|
.all()
|
|
)
|
|
for link in links:
|
|
links_by_identity.setdefault(link.identity_id, []).append(link)
|
|
|
|
return IdentityListResponse(
|
|
identities=[
|
|
_identity_item(identity, links_by_identity.get(identity.id, []))
|
|
for identity in identities
|
|
]
|
|
)
|
|
|
|
|
|
def _identity_item(identity: Identity, links: list[IdentityAccountLink]) -> IdentityItem:
|
|
primary_link = next((link for link in links if link.is_primary), None)
|
|
return IdentityItem(
|
|
id=identity.id,
|
|
display_name=identity.display_name,
|
|
external_subject=identity.external_subject,
|
|
source=identity.source,
|
|
primary_account_id=primary_link.account_id if primary_link is not None else None,
|
|
account_ids=[link.account_id for link in links],
|
|
status="active" if identity.is_active else "inactive",
|
|
)
|