107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.postbox import (
|
|
PostboxPortalEntryRef,
|
|
PostboxPortalProjectionProvider,
|
|
)
|
|
from govoplan_core.core.modules import ModuleContext
|
|
from govoplan_postbox.backend.db.models import Postbox, PostboxMessage
|
|
from govoplan_postbox.backend.principals import actor_from_principal
|
|
from govoplan_postbox.backend.runtime import configure_runtime, get_service
|
|
|
|
|
|
class PortalProjection(PostboxPortalProjectionProvider):
|
|
"""Read-only Portal projection; Postbox remains the access authority."""
|
|
|
|
def list_portal_entries(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
limit: int = 100,
|
|
) -> tuple[PostboxPortalEntryRef, ...]:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Postbox Portal projection requires a SQLAlchemy session.")
|
|
if not isinstance(principal, ApiPrincipal):
|
|
raise TypeError("Postbox Portal projection requires an API principal.")
|
|
if principal.tenant_id != tenant_id:
|
|
return ()
|
|
actor = actor_from_principal(principal)
|
|
visible = tuple(
|
|
get_service().list_visible_postboxes(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
actor=actor,
|
|
)
|
|
)
|
|
if not visible:
|
|
return ()
|
|
visible_by_id = {entry.id: entry for entry in visible}
|
|
rows = (
|
|
session.query(Postbox)
|
|
.filter(
|
|
Postbox.tenant_id == tenant_id,
|
|
Postbox.id.in_(tuple(visible_by_id)),
|
|
Postbox.status == "active",
|
|
)
|
|
.all()
|
|
)
|
|
enabled_ids = {
|
|
row.id
|
|
for row in rows
|
|
if bool((row.settings or {}).get("portal_visible"))
|
|
}
|
|
if not enabled_ids:
|
|
return ()
|
|
ordered_ids = tuple(
|
|
entry.id
|
|
for entry in sorted(
|
|
(visible_by_id[item_id] for item_id in enabled_ids),
|
|
key=lambda item: (item.name.casefold(), item.id),
|
|
)[: max(1, min(limit, 500))]
|
|
)
|
|
counts = get_service().message_counts_by_postbox(
|
|
session,
|
|
tenant_id=tenant_id,
|
|
postbox_ids=ordered_ids,
|
|
actor=actor,
|
|
)
|
|
latest_rows = (
|
|
session.query(
|
|
PostboxMessage.postbox_id,
|
|
func.max(PostboxMessage.delivered_at),
|
|
)
|
|
.filter(
|
|
PostboxMessage.tenant_id == tenant_id,
|
|
PostboxMessage.postbox_id.in_(ordered_ids),
|
|
PostboxMessage.classification.in_(
|
|
tuple(actor.authorized_classifications)
|
|
),
|
|
)
|
|
.group_by(PostboxMessage.postbox_id)
|
|
.all()
|
|
)
|
|
latest = {str(postbox_id): delivered_at for postbox_id, delivered_at in latest_rows}
|
|
return tuple(
|
|
PostboxPortalEntryRef(
|
|
postbox=visible_by_id[postbox_id],
|
|
unread_count=int(counts.get(postbox_id, {}).get("unread", 0)),
|
|
latest_message_at=latest.get(postbox_id),
|
|
route_path=f"/postbox?postbox={postbox_id}",
|
|
)
|
|
for postbox_id in ordered_ids
|
|
)
|
|
|
|
|
|
def create_portal_projection(context: ModuleContext) -> PortalProjection:
|
|
configure_runtime(registry=context.registry)
|
|
return PortalProjection()
|
|
|
|
|
|
__all__ = ["PortalProjection", "create_portal_projection"]
|