517 lines
17 KiB
Python
517 lines
17 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from collections.abc import Mapping, Sequence
|
|
from datetime import UTC, datetime
|
|
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from govoplan_core.core.change_sequence import record_change
|
|
from govoplan_core.core.concurrency import (
|
|
RevisionConflictError,
|
|
claim_revision,
|
|
strong_resource_etag,
|
|
)
|
|
from govoplan_core.core.idm import CAPABILITY_IDM_DIRECTORY, IdmDirectory
|
|
from govoplan_core.core.tasks import (
|
|
TaskCreateCommand,
|
|
WorkAssignmentRef,
|
|
WorkItem,
|
|
WorkItemPage,
|
|
WorkItemQuery,
|
|
WorkSourceRef,
|
|
)
|
|
from govoplan_tasks.backend.db.models import TaskAssignment, TaskItem
|
|
|
|
|
|
READ_SCOPE = "tasks:item:read"
|
|
WRITE_SCOPE = "tasks:item:write"
|
|
ADMIN_SCOPE = "tasks:item:admin"
|
|
PROVIDER_ID = "tasks.explicit"
|
|
ACTIVE_STATUSES = ("open", "in_progress", "deferred", "blocked")
|
|
|
|
|
|
class TaskError(RuntimeError):
|
|
code = "task_error"
|
|
|
|
|
|
class TaskNotFound(TaskError):
|
|
code = "task_not_found"
|
|
|
|
|
|
class TaskForbidden(TaskError):
|
|
code = "task_forbidden"
|
|
|
|
|
|
class TaskConflict(TaskError):
|
|
code = "task_conflict"
|
|
|
|
|
|
class SqlTaskService:
|
|
def __init__(self, registry: object | None = None) -> None:
|
|
self.registry = registry
|
|
|
|
def list_items(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
query: WorkItemQuery,
|
|
) -> WorkItemPage:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Tasks requires a SQLAlchemy session.")
|
|
self._require_tenant(principal, query.tenant_id)
|
|
if not _has(principal, READ_SCOPE) and not _has(principal, ADMIN_SCOPE):
|
|
raise TaskForbidden("The current principal may not read tasks.")
|
|
statement = self._visible_statement(principal, query.tenant_id)
|
|
if query.statuses:
|
|
statement = statement.where(TaskItem.status.in_(query.statuses))
|
|
if query.priorities:
|
|
statement = statement.where(TaskItem.priority.in_(query.priorities))
|
|
if query.due_before is not None:
|
|
statement = statement.where(TaskItem.due_at <= query.due_before)
|
|
if query.text:
|
|
pattern = f"%{_escape_like(query.text)}%"
|
|
statement = statement.where(
|
|
or_(
|
|
TaskItem.title.ilike(pattern, escape="\\"),
|
|
TaskItem.summary.ilike(pattern, escape="\\"),
|
|
TaskItem.required_action.ilike(pattern, escape="\\"),
|
|
)
|
|
)
|
|
count_statement = select(func.count()).select_from(
|
|
statement.order_by(None).subquery()
|
|
)
|
|
total = int(session.scalar(count_statement) or 0)
|
|
rows = list(
|
|
session.scalars(
|
|
statement.order_by(
|
|
_priority_rank(),
|
|
TaskItem.due_at.is_(None),
|
|
TaskItem.due_at.asc(),
|
|
TaskItem.updated_at.desc(),
|
|
TaskItem.id.desc(),
|
|
).limit(query.limit)
|
|
).unique()
|
|
)
|
|
return WorkItemPage(
|
|
items=tuple(self.to_item(row) for row in rows),
|
|
total=total,
|
|
truncated=total > len(rows),
|
|
)
|
|
|
|
def get_task(
|
|
self,
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
task_id: str,
|
|
for_update: bool = False,
|
|
) -> TaskItem:
|
|
self._require_tenant(principal, tenant_id)
|
|
statement = self._visible_statement(principal, tenant_id).where(
|
|
TaskItem.id == task_id
|
|
)
|
|
if for_update:
|
|
statement = statement.with_for_update()
|
|
task = session.scalar(statement)
|
|
if task is None:
|
|
raise TaskNotFound("Task not found or not visible.")
|
|
return task
|
|
|
|
def create_task(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
command: TaskCreateCommand,
|
|
) -> WorkItem:
|
|
if not isinstance(session, Session):
|
|
raise TypeError("Tasks requires a SQLAlchemy session.")
|
|
self._require_tenant(principal, command.tenant_id)
|
|
if not _has(principal, WRITE_SCOPE) and not _has(principal, ADMIN_SCOPE):
|
|
raise TaskForbidden("The current principal may not create tasks.")
|
|
digest = _command_digest(command)
|
|
existing = session.scalar(
|
|
select(TaskItem)
|
|
.where(
|
|
TaskItem.tenant_id == command.tenant_id,
|
|
TaskItem.idempotency_key == command.idempotency_key,
|
|
)
|
|
.options(selectinload(TaskItem.assignments))
|
|
)
|
|
if existing is not None:
|
|
if existing.request_sha256 != digest:
|
|
raise TaskConflict(
|
|
"The idempotency key already identifies a different task request."
|
|
)
|
|
return self.to_item(existing)
|
|
|
|
sources = [_source_dict(item) for item in command.sources]
|
|
primary = command.sources[0] if command.sources else None
|
|
actor_id = _account_id(principal)
|
|
task = TaskItem(
|
|
tenant_id=command.tenant_id,
|
|
title=command.title.strip(),
|
|
summary=_optional(command.summary),
|
|
status="open",
|
|
priority=command.priority,
|
|
due_at=command.due_at,
|
|
required_action=_optional(command.required_action),
|
|
action_url=_optional(command.action_url),
|
|
source_module=primary.module_id if primary else None,
|
|
source_resource_type=primary.resource_type if primary else None,
|
|
source_resource_id=primary.resource_id if primary else None,
|
|
source_revision=primary.revision if primary else None,
|
|
sources=sources,
|
|
provenance=dict(command.provenance),
|
|
metadata_=dict(command.metadata),
|
|
idempotency_key=command.idempotency_key.strip(),
|
|
request_sha256=digest,
|
|
created_by=actor_id,
|
|
updated_by=actor_id,
|
|
)
|
|
task.assignments = [
|
|
TaskAssignment(
|
|
tenant_id=command.tenant_id,
|
|
assignment_kind=item.kind,
|
|
assignment_id=item.id,
|
|
assignment_label=item.label,
|
|
)
|
|
for item in _deduplicate_assignments(command.assignments)
|
|
]
|
|
session.add(task)
|
|
session.flush()
|
|
record_change(
|
|
session,
|
|
module_id="tasks",
|
|
collection="work_items",
|
|
resource_type="task",
|
|
resource_id=task.id,
|
|
operation="created",
|
|
tenant_id=task.tenant_id,
|
|
actor_type="account" if actor_id else None,
|
|
actor_id=actor_id,
|
|
payload={
|
|
"status": task.status,
|
|
"priority": task.priority,
|
|
"assignment_count": len(task.assignments),
|
|
"source_module": task.source_module,
|
|
},
|
|
)
|
|
return self.to_item(task)
|
|
|
|
def transition_task(
|
|
self,
|
|
session: Session,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
task_id: str,
|
|
action: str,
|
|
expected_revision: int,
|
|
deferred_until: datetime | None = None,
|
|
comment: str | None = None,
|
|
) -> WorkItem:
|
|
if not _has(principal, WRITE_SCOPE) and not _has(principal, ADMIN_SCOPE):
|
|
raise TaskForbidden("The current principal may not update tasks.")
|
|
task = self.get_task(
|
|
session,
|
|
principal,
|
|
tenant_id=tenant_id,
|
|
task_id=task_id,
|
|
for_update=True,
|
|
)
|
|
previous_status = task.status
|
|
next_status = _next_status(task.status, action)
|
|
if action == "defer":
|
|
if deferred_until is None or _utc(deferred_until) <= datetime.now(UTC):
|
|
raise TaskConflict("Deferred tasks require a future date and time.")
|
|
try:
|
|
next_revision = claim_revision(
|
|
session,
|
|
model=TaskItem,
|
|
filters=(TaskItem.id == task.id, TaskItem.tenant_id == tenant_id),
|
|
revision_attribute="revision",
|
|
expected_revision=expected_revision,
|
|
resource_type="task",
|
|
resource_id=task.id,
|
|
refresh_path=f"/api/v1/tasks/{task.id}",
|
|
)
|
|
except RevisionConflictError:
|
|
raise
|
|
session.refresh(task)
|
|
actor_id = _account_id(principal)
|
|
now = datetime.now(UTC)
|
|
task.revision = next_revision
|
|
task.status = next_status
|
|
task.updated_by = actor_id
|
|
task.deferred_until = _utc(deferred_until) if action == "defer" else None
|
|
if action == "complete":
|
|
task.completed_at = now
|
|
task.completed_by = actor_id
|
|
elif action == "reopen":
|
|
task.completed_at = None
|
|
task.completed_by = None
|
|
task.cancelled_at = None
|
|
elif action == "cancel":
|
|
task.cancelled_at = now
|
|
metadata = dict(task.metadata_ or {})
|
|
history = list(metadata.get("transition_history") or [])
|
|
history.append(
|
|
{
|
|
"action": action,
|
|
"from_status": previous_status,
|
|
"to_status": next_status,
|
|
"actor_id": actor_id,
|
|
"recorded_at": now.isoformat(),
|
|
"comment": _optional(comment),
|
|
}
|
|
)
|
|
metadata["transition_history"] = history[-100:]
|
|
task.metadata_ = metadata
|
|
session.flush()
|
|
record_change(
|
|
session,
|
|
module_id="tasks",
|
|
collection="work_items",
|
|
resource_type="task",
|
|
resource_id=task.id,
|
|
operation="updated",
|
|
tenant_id=tenant_id,
|
|
actor_type="account" if actor_id else None,
|
|
actor_id=actor_id,
|
|
payload={
|
|
"action": action,
|
|
"status": next_status,
|
|
"revision": next_revision,
|
|
},
|
|
)
|
|
return self.to_item(task)
|
|
|
|
def _visible_statement(self, principal: object, tenant_id: str):
|
|
statement = (
|
|
select(TaskItem)
|
|
.where(TaskItem.tenant_id == tenant_id)
|
|
.options(selectinload(TaskItem.assignments))
|
|
)
|
|
if _has(principal, ADMIN_SCOPE):
|
|
return statement
|
|
targets = self._assignment_targets(principal, tenant_id)
|
|
conditions = [
|
|
and_(
|
|
TaskAssignment.assignment_kind == kind,
|
|
TaskAssignment.assignment_id.in_(tuple(ids)),
|
|
)
|
|
for kind, ids in targets.items()
|
|
if ids
|
|
]
|
|
if not conditions:
|
|
return statement.where(False)
|
|
return statement.join(TaskAssignment).where(or_(*conditions)).distinct()
|
|
|
|
def _assignment_targets(
|
|
self, principal: object, tenant_id: str
|
|
) -> dict[str, set[str]]:
|
|
targets = {
|
|
"account": {_account_id(principal)} if _account_id(principal) else set(),
|
|
"group": set(getattr(principal, "group_ids", ()) or ()),
|
|
"role": set(getattr(principal, "role_ids", ()) or ()),
|
|
"function_assignment": set(
|
|
getattr(principal, "function_assignment_ids", ()) or ()
|
|
),
|
|
"function": set(),
|
|
"anyone": {"*"},
|
|
}
|
|
directory = self._idm_directory()
|
|
if directory is not None and _account_id(principal):
|
|
assignments = directory.organization_function_assignments_for_account(
|
|
_account_id(principal),
|
|
tenant_id=tenant_id,
|
|
)
|
|
targets["function"].update(
|
|
item.function_id
|
|
for item in assignments
|
|
if item.status == "active" and item.tenant_id == tenant_id
|
|
)
|
|
return targets
|
|
|
|
def _idm_directory(self) -> IdmDirectory | None:
|
|
registry = self.registry
|
|
if (
|
|
registry is None
|
|
or not hasattr(registry, "has_capability")
|
|
or not registry.has_capability(CAPABILITY_IDM_DIRECTORY)
|
|
):
|
|
return None
|
|
provider = registry.capability(CAPABILITY_IDM_DIRECTORY)
|
|
return provider if isinstance(provider, IdmDirectory) else None
|
|
|
|
@staticmethod
|
|
def _require_tenant(principal: object, tenant_id: str) -> None:
|
|
if str(getattr(principal, "tenant_id", "") or "") != tenant_id:
|
|
raise TaskForbidden("Task access is limited to the active tenant.")
|
|
|
|
@staticmethod
|
|
def to_item(task: TaskItem) -> WorkItem:
|
|
return WorkItem(
|
|
id=task.id,
|
|
provider_id=PROVIDER_ID,
|
|
owner_module="tasks",
|
|
tenant_id=task.tenant_id,
|
|
title=task.title,
|
|
summary=task.summary,
|
|
status=task.status, # type: ignore[arg-type]
|
|
priority=task.priority, # type: ignore[arg-type]
|
|
required_action=task.required_action,
|
|
action_url=task.action_url,
|
|
due_at=task.due_at,
|
|
deferred_until=task.deferred_until,
|
|
assignments=tuple(
|
|
WorkAssignmentRef(
|
|
kind=item.assignment_kind, # type: ignore[arg-type]
|
|
id=item.assignment_id,
|
|
label=item.assignment_label,
|
|
)
|
|
for item in task.assignments
|
|
),
|
|
sources=tuple(WorkSourceRef(**item) for item in task.sources),
|
|
provenance=dict(task.provenance or {}),
|
|
metadata=dict(task.metadata_ or {}),
|
|
revision=str(task.revision),
|
|
created_at=task.created_at,
|
|
updated_at=task.updated_at,
|
|
)
|
|
|
|
|
|
def task_etag(task: WorkItem) -> str | None:
|
|
try:
|
|
return strong_resource_etag("task", task.id, int(task.revision))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def _has(principal: object, scope: str) -> bool:
|
|
checker = getattr(principal, "has", None)
|
|
return bool(callable(checker) and checker(scope))
|
|
|
|
|
|
def _account_id(principal: object) -> str:
|
|
return str(getattr(principal, "account_id", "") or "")
|
|
|
|
|
|
def _optional(value: str | None) -> str | None:
|
|
normalized = str(value or "").strip()
|
|
return normalized or None
|
|
|
|
|
|
def _utc(value: datetime) -> datetime:
|
|
if value.tzinfo is None:
|
|
return value.replace(tzinfo=UTC)
|
|
return value.astimezone(UTC)
|
|
|
|
|
|
def _source_dict(value: WorkSourceRef) -> dict[str, str | None]:
|
|
return {
|
|
"module_id": value.module_id,
|
|
"resource_type": value.resource_type,
|
|
"resource_id": value.resource_id,
|
|
"revision": value.revision,
|
|
"url": value.url,
|
|
"label": value.label,
|
|
}
|
|
|
|
|
|
def _deduplicate_assignments(
|
|
assignments: Sequence[WorkAssignmentRef],
|
|
) -> tuple[WorkAssignmentRef, ...]:
|
|
by_key: dict[tuple[str, str], WorkAssignmentRef] = {}
|
|
for item in assignments:
|
|
by_key.setdefault((item.kind, item.id), item)
|
|
return tuple(by_key.values())
|
|
|
|
|
|
def _command_digest(command: TaskCreateCommand) -> str:
|
|
payload = {
|
|
"tenant_id": command.tenant_id,
|
|
"title": command.title.strip(),
|
|
"summary": _optional(command.summary),
|
|
"priority": command.priority,
|
|
"due_at": command.due_at.isoformat() if command.due_at else None,
|
|
"required_action": _optional(command.required_action),
|
|
"action_url": _optional(command.action_url),
|
|
"assignments": [
|
|
{"kind": item.kind, "id": item.id, "label": item.label}
|
|
for item in _deduplicate_assignments(command.assignments)
|
|
],
|
|
"sources": [_source_dict(item) for item in command.sources],
|
|
"provenance": dict(command.provenance),
|
|
"metadata": dict(command.metadata),
|
|
}
|
|
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str)
|
|
return hashlib.sha256(encoded.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _next_status(current: str, action: str) -> str:
|
|
allowed: Mapping[str, Mapping[str, str]] = {
|
|
"open": {
|
|
"start": "in_progress",
|
|
"complete": "completed",
|
|
"defer": "deferred",
|
|
"cancel": "cancelled",
|
|
},
|
|
"in_progress": {
|
|
"complete": "completed",
|
|
"defer": "deferred",
|
|
"cancel": "cancelled",
|
|
},
|
|
"deferred": {
|
|
"start": "in_progress",
|
|
"complete": "completed",
|
|
"reopen": "open",
|
|
"cancel": "cancelled",
|
|
},
|
|
"blocked": {
|
|
"reopen": "open",
|
|
"cancel": "cancelled",
|
|
},
|
|
"completed": {"reopen": "open"},
|
|
"cancelled": {"reopen": "open"},
|
|
}
|
|
next_status = allowed.get(current, {}).get(action)
|
|
if next_status is None:
|
|
raise TaskConflict(
|
|
f"Action {action!r} is not available for a {current!r} task."
|
|
)
|
|
return next_status
|
|
|
|
|
|
def _escape_like(value: str) -> str:
|
|
return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
|
|
|
|
def _priority_rank():
|
|
from sqlalchemy import case
|
|
|
|
return case(
|
|
(TaskItem.priority == "urgent", 0),
|
|
(TaskItem.priority == "high", 1),
|
|
(TaskItem.priority == "normal", 2),
|
|
else_=3,
|
|
)
|
|
|
|
|
|
__all__ = [
|
|
"ACTIVE_STATUSES",
|
|
"PROVIDER_ID",
|
|
"SqlTaskService",
|
|
"TaskConflict",
|
|
"TaskError",
|
|
"TaskForbidden",
|
|
"TaskNotFound",
|
|
"task_etag",
|
|
]
|