Skip to content

donna.skills.tools.task_db_read

donna.skills.tools.task_db_read

task_db_read — thin read-only wrapper around Database.get_task / list_tasks.

Registered into DEFAULT_TOOL_REGISTRY via donna.skills.tools.register_default_tools. Only registered when a Database handle is available at boot.

Read-only by construction: never imports or references create/update/transition.

logger module-attribute 

logger = get_logger()

TaskDbReadError

Bases: Exception

Raised when a task_db_read invocation fails.

task_db_read async 

task_db_read(*, client: Any, task_id: str | None = None, user_id: str | None = None, status: str | None = None, domain: str | None = None) -> dict[str, Any]

Fetch one task (by task_id) or a filtered list.

When task_id is provided, returns {ok, task}; otherwise returns {ok, tasks} filtered by any of user_id / status / domain.

Source code in src/donna/skills/tools/task_db_read.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
async def task_db_read(
    *,
    client: Any,
    task_id: str | None = None,
    user_id: str | None = None,
    status: str | None = None,
    domain: str | None = None,
) -> dict[str, Any]:
    """Fetch one task (by ``task_id``) or a filtered list.

    When ``task_id`` is provided, returns ``{ok, task}``; otherwise returns
    ``{ok, tasks}`` filtered by any of ``user_id`` / ``status`` / ``domain``.
    """
    if task_id is not None:
        if not task_id.strip():
            raise TaskDbReadError("task_id must be non-empty")
        if user_id is not None or status is not None or domain is not None:
            raise TaskDbReadError(
                "task_id is mutually exclusive with user_id/status/domain filters"
            )
        try:
            row = await client.get_task(task_id)
        except Exception as exc:
            logger.warning("task_db_read_get_failed", task_id=task_id, error=str(exc))
            raise TaskDbReadError(f"get_task: {exc}") from exc
        if row is None:
            raise TaskDbReadError(f"task not found: {task_id}")
        return {"ok": True, "task": _project(row)}

    status_enum = _coerce_status(status) if status is not None else None
    domain_enum = _coerce_domain(domain) if domain is not None else None
    try:
        rows = await client.list_tasks(
            user_id=user_id, status=status_enum, domain=domain_enum,
        )
    except Exception as exc:
        logger.warning(
            "task_db_read_list_failed",
            user_id=user_id, status=status, domain=domain, error=str(exc),
        )
        raise TaskDbReadError(f"list_tasks: {exc}") from exc
    return {"ok": True, "tasks": [_project(r) for r in rows]}