263 lines
7.9 KiB
Python
263 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Sequence
|
|
|
|
from sqlalchemy import func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account, Group, User
|
|
from govoplan_core.core.references import (
|
|
ReferenceOption,
|
|
ReferenceSearchPage,
|
|
ReferenceSearchRequest,
|
|
)
|
|
|
|
|
|
class SqlAccessReferenceOptionProvider:
|
|
"""Principal-aware, bounded Access directory search."""
|
|
|
|
def search_reference_options(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ReferenceSearchRequest,
|
|
) -> ReferenceSearchPage:
|
|
db = _session(session)
|
|
tenant_id = str(request.tenant_id or "").strip()
|
|
if not tenant_id:
|
|
return ReferenceSearchPage()
|
|
limit = max(1, min(int(request.limit), 200))
|
|
offset = _cursor_offset(request.cursor)
|
|
selected = tuple(
|
|
dict.fromkeys(
|
|
str(value).strip()
|
|
for value in request.selected_values
|
|
if str(value).strip()
|
|
)
|
|
)[:200]
|
|
administrative = request.context.get("administrative") is True
|
|
query = str(request.query or "").strip().casefold()
|
|
if request.kind in {"user", "membership"}:
|
|
return _search_users(
|
|
db,
|
|
principal,
|
|
tenant_id=tenant_id,
|
|
kind=request.kind,
|
|
query=query,
|
|
selected=selected,
|
|
limit=limit,
|
|
offset=offset,
|
|
administrative=administrative,
|
|
)
|
|
if request.kind == "group":
|
|
return _search_groups(
|
|
db,
|
|
principal,
|
|
tenant_id=tenant_id,
|
|
query=query,
|
|
selected=selected,
|
|
limit=limit,
|
|
offset=offset,
|
|
administrative=administrative,
|
|
)
|
|
raise ValueError(f"Unsupported Access reference kind: {request.kind}")
|
|
|
|
|
|
def _search_users(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
kind: str,
|
|
query: str,
|
|
selected: Sequence[str],
|
|
limit: int,
|
|
offset: int,
|
|
administrative: bool,
|
|
) -> ReferenceSearchPage:
|
|
value_column = User.id if kind == "membership" else User.account_id
|
|
base = (
|
|
session.query(User, Account)
|
|
.join(Account, Account.id == User.account_id)
|
|
.filter(User.tenant_id == tenant_id)
|
|
)
|
|
if not administrative:
|
|
account_id = str(getattr(principal, "account_id", "") or "")
|
|
if not account_id:
|
|
return ReferenceSearchPage()
|
|
base = base.filter(User.account_id == account_id)
|
|
|
|
selected_rows = (
|
|
base.filter(value_column.in_(selected)).all()
|
|
if selected
|
|
else []
|
|
)
|
|
search_query = base
|
|
if selected:
|
|
search_query = search_query.filter(value_column.notin_(selected))
|
|
if query:
|
|
search_query = search_query.filter(
|
|
or_(
|
|
func.lower(func.coalesce(User.display_name, "")).contains(
|
|
query,
|
|
autoescape=True,
|
|
),
|
|
func.lower(User.email).contains(query, autoescape=True),
|
|
func.lower(Account.email).contains(query, autoescape=True),
|
|
func.lower(value_column).contains(query, autoescape=True),
|
|
)
|
|
)
|
|
rows = (
|
|
search_query.order_by(
|
|
func.lower(func.coalesce(User.display_name, User.email)).asc(),
|
|
value_column.asc(),
|
|
)
|
|
.offset(offset)
|
|
.limit(limit + 1)
|
|
.all()
|
|
)
|
|
has_more = len(rows) > limit
|
|
options = [
|
|
_user_option(user, account, kind=kind)
|
|
for user, account in rows[:limit]
|
|
]
|
|
selected_by_value = {
|
|
_user_value(user, kind=kind): _user_option(user, account, kind=kind)
|
|
for user, account in selected_rows
|
|
}
|
|
options.extend(
|
|
selected_by_value[value]
|
|
for value in selected
|
|
if value in selected_by_value
|
|
)
|
|
return ReferenceSearchPage(
|
|
options=tuple(options),
|
|
next_cursor=f"offset:{offset + limit}" if has_more else None,
|
|
has_more=has_more,
|
|
)
|
|
|
|
|
|
def _search_groups(
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
query: str,
|
|
selected: Sequence[str],
|
|
limit: int,
|
|
offset: int,
|
|
administrative: bool,
|
|
) -> ReferenceSearchPage:
|
|
base = session.query(Group).filter(Group.tenant_id == tenant_id)
|
|
if not administrative:
|
|
permitted = tuple(
|
|
dict.fromkeys(
|
|
str(group_id)
|
|
for group_id in getattr(principal, "group_ids", ())
|
|
if str(group_id)
|
|
)
|
|
)
|
|
if not permitted:
|
|
return ReferenceSearchPage()
|
|
base = base.filter(Group.id.in_(permitted))
|
|
|
|
selected_rows = base.filter(Group.id.in_(selected)).all() if selected else []
|
|
search_query = base
|
|
if selected:
|
|
search_query = search_query.filter(Group.id.notin_(selected))
|
|
if query:
|
|
search_query = search_query.filter(
|
|
or_(
|
|
func.lower(Group.name).contains(query, autoescape=True),
|
|
func.lower(Group.slug).contains(query, autoescape=True),
|
|
func.lower(Group.id).contains(query, autoescape=True),
|
|
)
|
|
)
|
|
rows = (
|
|
search_query.order_by(func.lower(Group.name).asc(), Group.id.asc())
|
|
.offset(offset)
|
|
.limit(limit + 1)
|
|
.all()
|
|
)
|
|
has_more = len(rows) > limit
|
|
options = [_group_option(group) for group in rows[:limit]]
|
|
selected_by_value = {group.id: _group_option(group) for group in selected_rows}
|
|
options.extend(
|
|
selected_by_value[value]
|
|
for value in selected
|
|
if value in selected_by_value
|
|
)
|
|
return ReferenceSearchPage(
|
|
options=tuple(options),
|
|
next_cursor=f"offset:{offset + limit}" if has_more else None,
|
|
has_more=has_more,
|
|
)
|
|
|
|
|
|
def _user_value(user: User, *, kind: str) -> str:
|
|
return user.id if kind == "membership" else user.account_id
|
|
|
|
|
|
def _user_option(user: User, account: Account, *, kind: str) -> ReferenceOption:
|
|
inactive = not user.is_active or not account.is_active
|
|
value = _user_value(user, kind=kind)
|
|
description_parts = [
|
|
user.email,
|
|
"Inactive" if inactive else None,
|
|
]
|
|
return ReferenceOption(
|
|
value=value,
|
|
label=user.display_name or user.email or value,
|
|
description=" · ".join(
|
|
part for part in description_parts if part
|
|
) or None,
|
|
kind=kind,
|
|
availability="inactive" if inactive else "available",
|
|
disabled=inactive,
|
|
source_module="access",
|
|
provenance={
|
|
"tenant_id": user.tenant_id,
|
|
"membership_id": user.id,
|
|
"account_id": user.account_id,
|
|
},
|
|
)
|
|
|
|
|
|
def _group_option(group: Group) -> ReferenceOption:
|
|
inactive = not group.is_active
|
|
return ReferenceOption(
|
|
value=group.id,
|
|
label=group.name or group.id,
|
|
description="Inactive" if inactive else None,
|
|
kind="group",
|
|
availability="inactive" if inactive else "available",
|
|
disabled=inactive,
|
|
source_module="access",
|
|
provenance={"tenant_id": group.tenant_id, "group_id": group.id},
|
|
)
|
|
|
|
|
|
def _cursor_offset(cursor: str | None) -> int:
|
|
if cursor is None:
|
|
return 0
|
|
prefix = "offset:"
|
|
if not cursor.startswith(prefix):
|
|
raise ValueError("Invalid reference search cursor.")
|
|
try:
|
|
offset = int(cursor[len(prefix):])
|
|
except ValueError as exc:
|
|
raise ValueError("Invalid reference search cursor.") from exc
|
|
if offset < 0:
|
|
raise ValueError("Invalid reference search cursor.")
|
|
return offset
|
|
|
|
|
|
def _session(session: object) -> Session:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Access reference search requires a SQLAlchemy Session")
|
|
return session
|
|
|
|
|
|
__all__ = ["SqlAccessReferenceOptionProvider"]
|