112 lines
4.2 KiB
Python
112 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Mapping, Sequence
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_access.backend.db.models import Account, ApiKey, Group, Role, User
|
|
from govoplan_core.core.access import AccessAdministration
|
|
|
|
|
|
class SqlAccessAdministration(AccessAdministration):
|
|
def tenant_counts(self, session: object, tenant_id: str) -> Mapping[str, int]:
|
|
db = _session(session)
|
|
return {
|
|
"users": db.query(User).filter(User.tenant_id == tenant_id).count(),
|
|
"active_users": db.query(User).filter(User.tenant_id == tenant_id, User.is_active.is_(True)).count(),
|
|
"groups": db.query(Group).filter(Group.tenant_id == tenant_id).count(),
|
|
"api_keys": db.query(ApiKey).filter(ApiKey.tenant_id == tenant_id).count(),
|
|
"active_api_keys": db.query(ApiKey).filter(ApiKey.tenant_id == tenant_id, ApiKey.revoked_at.is_(None)).count(),
|
|
}
|
|
|
|
def system_account_count(self, session: object) -> int:
|
|
db = _session(session)
|
|
return db.query(Account).count()
|
|
|
|
def role_count_for_tenant(self, session: object, tenant_id: str) -> int:
|
|
db = _session(session)
|
|
return db.query(Role).filter(Role.tenant_id == tenant_id).count()
|
|
|
|
def active_api_key_count_for_tenant(self, session: object, tenant_id: str) -> int:
|
|
db = _session(session)
|
|
return db.query(ApiKey).filter(ApiKey.tenant_id == tenant_id, ApiKey.revoked_at.is_(None)).count()
|
|
|
|
def actor_email_by_user_id(self, session: object, user_ids: Iterable[str]) -> Mapping[str, str | None]:
|
|
ids = {str(user_id) for user_id in user_ids if user_id}
|
|
if not ids:
|
|
return {}
|
|
db = _session(session)
|
|
rows = (
|
|
db.query(User.id, Account.email)
|
|
.join(Account, Account.id == User.account_id)
|
|
.filter(User.id.in_(ids))
|
|
.all()
|
|
)
|
|
return {user_id: email for user_id, email in rows}
|
|
|
|
def user_ids_for_actor_filter(self, session: object, *, operator: str, value: str) -> Sequence[str]:
|
|
normalized = value.casefold().strip()
|
|
if not normalized:
|
|
return ()
|
|
db = _session(session)
|
|
text = func.lower(func.coalesce(Account.email, ""))
|
|
condition = text == normalized if operator == "eq" else text.contains(normalized)
|
|
rows = (
|
|
db.query(User.id)
|
|
.join(Account, Account.id == User.account_id)
|
|
.filter(condition)
|
|
.all()
|
|
)
|
|
return tuple(user_id for (user_id,) in rows)
|
|
|
|
def user_settings(self, session: object, user_id: str, *, tenant_id: str) -> Mapping[str, object] | None:
|
|
user = _session(session).get(User, user_id)
|
|
if user is None or user.tenant_id != tenant_id:
|
|
return None
|
|
return dict(user.settings or {})
|
|
|
|
def set_user_settings(
|
|
self,
|
|
session: object,
|
|
user_id: str,
|
|
*,
|
|
tenant_id: str,
|
|
settings: Mapping[str, object],
|
|
) -> Mapping[str, object] | None:
|
|
db = _session(session)
|
|
user = db.get(User, user_id)
|
|
if user is None or user.tenant_id != tenant_id:
|
|
return None
|
|
user.settings = dict(settings)
|
|
db.add(user)
|
|
return dict(user.settings or {})
|
|
|
|
def group_settings(self, session: object, group_id: str, *, tenant_id: str) -> Mapping[str, object] | None:
|
|
group = _session(session).get(Group, group_id)
|
|
if group is None or group.tenant_id != tenant_id:
|
|
return None
|
|
return dict(group.settings or {})
|
|
|
|
def set_group_settings(
|
|
self,
|
|
session: object,
|
|
group_id: str,
|
|
*,
|
|
tenant_id: str,
|
|
settings: Mapping[str, object],
|
|
) -> Mapping[str, object] | None:
|
|
db = _session(session)
|
|
group = db.get(Group, group_id)
|
|
if group is None or group.tenant_id != tenant_id:
|
|
return None
|
|
group.settings = dict(settings)
|
|
db.add(group)
|
|
return dict(group.settings or {})
|
|
|
|
|
|
def _session(session: object) -> Session:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Access administration requires a SQLAlchemy Session")
|
|
return session
|