168 lines
6.5 KiB
Python
168 lines
6.5 KiB
Python
# [DEF:TaskContextModule:Module]
|
|
# @SEMANTICS: task, context, plugin, execution, logger
|
|
# @PURPOSE: Provides execution context passed to plugins during task execution.
|
|
# @LAYER: Core
|
|
# @RELATION: DEPENDS_ON -> [TaskLoggerModule]
|
|
# @RELATION: DEPENDS_ON -> [TaskManager]
|
|
# @COMPLEXITY: 5
|
|
# @INVARIANT: Each TaskContext is bound to a single task execution.
|
|
# @PRE: Task execution pipeline provides valid task identifiers, logging callbacks, and parameter dictionaries.
|
|
# @POST: Plugins receive context instances with stable logger and parameter accessors.
|
|
# @SIDE_EFFECT: Creates task-scoped logger wrappers and carries optional background task handles across sub-contexts.
|
|
# @DATA_CONTRACT: Input[task_id, add_log_fn, params, default_source, background_tasks] -> Output[TaskContext]
|
|
|
|
# [SECTION: IMPORTS]
|
|
# [SECTION: IMPORTS]
|
|
from typing import Dict, Any, Callable, Optional
|
|
from .task_logger import TaskLogger
|
|
from ..logger import belief_scope
|
|
# [/SECTION]
|
|
|
|
|
|
# [DEF:TaskContext:Class]
|
|
# @SEMANTICS: context, task, execution, plugin
|
|
# @PURPOSE: A container passed to plugin.execute() providing the logger and other task-specific utilities.
|
|
# @COMPLEXITY: 5
|
|
# @INVARIANT: logger is always a valid TaskLogger instance.
|
|
# @PRE: Constructor receives non-empty task_id, callable add_log_fn, and params mapping.
|
|
# @POST: Instance exposes immutable task identity with logger, params, and optional background task access.
|
|
# @RELATION: DEPENDS_ON -> [TaskLogger]
|
|
# @SIDE_EFFECT: Emits structured task logs through TaskLogger on plugin interactions.
|
|
# @DATA_CONTRACT: Input[task_id, add_log_fn, params, default_source, background_tasks] -> Output[TaskContext]
|
|
# @UX_STATE: Idle -> Active -> Complete
|
|
#
|
|
# @TEST_CONTRACT: TaskContextContract ->
|
|
# {
|
|
# required_fields: {task_id: str, add_log_fn: Callable, params: dict},
|
|
# optional_fields: {default_source: str},
|
|
# invariants: [
|
|
# "task_id matches initialized logger's task_id",
|
|
# "logger is a valid TaskLogger instance"
|
|
# ]
|
|
# }
|
|
# @TEST_FIXTURE: valid_context -> {"task_id": "123", "add_log_fn": lambda *args: None, "params": {"k": "v"}, "default_source": "plugin"}
|
|
# @TEST_EDGE: missing_task_id -> raises TypeError
|
|
# @TEST_EDGE: missing_add_log_fn -> raises TypeError
|
|
# @TEST_INVARIANT: logger_initialized -> verifies: [valid_context]
|
|
class TaskContext:
|
|
"""
|
|
Execution context provided to plugins during task execution.
|
|
|
|
Usage:
|
|
def execute(params: dict, context: TaskContext = None):
|
|
if context:
|
|
context.logger.info("Starting process")
|
|
context.logger.progress("Processing items", percent=50)
|
|
# ... plugin logic
|
|
"""
|
|
|
|
# [DEF:__init__:Function]
|
|
# @PURPOSE: Initialize the TaskContext with task-specific resources.
|
|
# @PRE: task_id is a valid task identifier, add_log_fn is callable.
|
|
# @POST: TaskContext is ready to be passed to plugin.execute().
|
|
# @PARAM: task_id (str) - The ID of the task.
|
|
# @PARAM: add_log_fn (Callable) - Function to add log to TaskManager.
|
|
# @PARAM: params (Dict) - Task parameters.
|
|
# @PARAM: default_source (str) - Default source for logs (default: "plugin").
|
|
def __init__(
|
|
self,
|
|
task_id: str,
|
|
add_log_fn: Callable,
|
|
params: Dict[str, Any],
|
|
default_source: str = "plugin",
|
|
background_tasks: Optional[Any] = None,
|
|
):
|
|
with belief_scope("__init__"):
|
|
self._task_id = task_id
|
|
self._params = params
|
|
self._background_tasks = background_tasks
|
|
self._logger = TaskLogger(
|
|
task_id=task_id, add_log_fn=add_log_fn, source=default_source
|
|
)
|
|
|
|
# [/DEF:__init__:Function]
|
|
|
|
# [DEF:task_id:Function]
|
|
# @PURPOSE: Get the task ID.
|
|
# @PRE: TaskContext must be initialized.
|
|
# @POST: Returns the task ID string.
|
|
# @RETURN: str - The task ID.
|
|
@property
|
|
def task_id(self) -> str:
|
|
with belief_scope("task_id"):
|
|
return self._task_id
|
|
|
|
# [/DEF:task_id:Function]
|
|
|
|
# [DEF:logger:Function]
|
|
# @PURPOSE: Get the TaskLogger instance for this context.
|
|
# @PRE: TaskContext must be initialized.
|
|
# @POST: Returns the TaskLogger instance.
|
|
# @RETURN: TaskLogger - The logger instance.
|
|
@property
|
|
def logger(self) -> TaskLogger:
|
|
with belief_scope("logger"):
|
|
return self._logger
|
|
|
|
# [/DEF:logger:Function]
|
|
|
|
# [DEF:params:Function]
|
|
# @PURPOSE: Get the task parameters.
|
|
# @PRE: TaskContext must be initialized.
|
|
# @POST: Returns the parameters dictionary.
|
|
# @RETURN: Dict[str, Any] - The task parameters.
|
|
@property
|
|
def params(self) -> Dict[str, Any]:
|
|
with belief_scope("params"):
|
|
return self._params
|
|
|
|
# [/DEF:params:Function]
|
|
|
|
# [DEF:background_tasks:Function]
|
|
# @PURPOSE: Expose optional background task scheduler for plugins that dispatch deferred side effects.
|
|
# @PRE: TaskContext must be initialized.
|
|
# @POST: Returns BackgroundTasks-like object or None.
|
|
@property
|
|
def background_tasks(self) -> Optional[Any]:
|
|
with belief_scope("background_tasks"):
|
|
return self._background_tasks
|
|
|
|
# [/DEF:background_tasks:Function]
|
|
|
|
# [DEF:get_param:Function]
|
|
# @PURPOSE: Get a specific parameter value with optional default.
|
|
# @PRE: TaskContext must be initialized.
|
|
# @POST: Returns parameter value or default.
|
|
# @PARAM: key (str) - Parameter key.
|
|
# @PARAM: default (Any) - Default value if key not found.
|
|
# @RETURN: Any - Parameter value or default.
|
|
def get_param(self, key: str, default: Any = None) -> Any:
|
|
with belief_scope("get_param"):
|
|
return self._params.get(key, default)
|
|
|
|
# [/DEF:get_param:Function]
|
|
|
|
# [DEF:create_sub_context:Function]
|
|
# @PURPOSE: Create a sub-context with a different default source.
|
|
# @PRE: source is a non-empty string.
|
|
# @POST: Returns new TaskContext with different logger source.
|
|
# @PARAM: source (str) - New default source for logging.
|
|
# @RETURN: TaskContext - New context with different source.
|
|
def create_sub_context(self, source: str) -> "TaskContext":
|
|
"""Create a sub-context with a different default source for logging."""
|
|
with belief_scope("create_sub_context"):
|
|
return TaskContext(
|
|
task_id=self._task_id,
|
|
add_log_fn=self._logger._add_log,
|
|
params=self._params,
|
|
default_source=source,
|
|
background_tasks=self._background_tasks,
|
|
)
|
|
|
|
# [/DEF:create_sub_context:Function]
|
|
|
|
|
|
# [/DEF:TaskContext:Class]
|
|
|
|
# [/DEF:TaskContextModule:Module]
|