File size: 4,434 Bytes
1425afc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 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 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | """
V11 Execution Context
---------------------
Central runtime object passed into every task.
Responsibilities:
- Hold inputs
- Share memory between tasks
- Store outputs
- Track execution metadata
- Provide filesystem helpers
- Provide logging helpers
"""
from __future__ import annotations
import os
import uuid
import tempfile
from typing import Any, Dict, Optional
# =========================================================
# Context Object
# =========================================================
class ExecutionContext:
"""
Standard runtime context used by ALL tasks.
Every task receives:
async def run(ctx: ExecutionContext)
"""
# -----------------------------------------------------
# INIT
# -----------------------------------------------------
def __init__(
self,
task_name: str,
inputs: Optional[Dict[str, Any]] = None,
workspace: Optional[str] = None,
):
self.task_name = task_name
self.job_id = str(uuid.uuid4())
self.inputs: Dict[str, Any] = inputs or {}
self.outputs: Dict[str, Any] = {}
self.memory: Dict[str, Any] = {}
self.status: str = "created"
self.error: Optional[str] = None
self.workspace = workspace or self._create_workspace()
# -----------------------------------------------------
# WORKSPACE
# -----------------------------------------------------
def _create_workspace(self) -> str:
path = tempfile.mkdtemp(prefix="basyx_job_")
return path
def path(self, filename: str) -> str:
"""
Safe workspace path helper
"""
return os.path.join(self.workspace, filename)
# -----------------------------------------------------
# INPUT HELPERS
# -----------------------------------------------------
def get(self, key: str, default=None):
return self.inputs.get(key, default)
def require(self, key: str):
if key not in self.inputs:
raise ValueError(f"Missing required input: {key}")
return self.inputs[key]
# -----------------------------------------------------
# OUTPUT HELPERS
# -----------------------------------------------------
def set_output(self, key: str, value: Any):
self.outputs[key] = value
def result(self) -> Dict[str, Any]:
return {
"job_id": self.job_id,
"task": self.task_name,
"status": self.status,
"outputs": self.outputs,
"error": self.error,
}
# -----------------------------------------------------
# MEMORY (cross-task sharing)
# -----------------------------------------------------
def remember(self, key: str, value: Any):
"""
Save value for downstream tasks.
"""
self.memory[key] = value
def recall(self, key: str, default=None):
return self.memory.get(key, default)
# -----------------------------------------------------
# STATUS MANAGEMENT
# -----------------------------------------------------
def mark_running(self):
self.status = "running"
def mark_complete(self):
self.status = "completed"
def mark_failed(self, error: Exception | str):
self.status = "failed"
self.error = str(error)
# -----------------------------------------------------
# LOGGING
# -----------------------------------------------------
def log(self, message: str):
print(f"[{self.task_name} | {self.job_id}] {message}")
# -----------------------------------------------------
# SERIALIZATION
# -----------------------------------------------------
def to_dict(self):
return {
"job_id": self.job_id,
"task_name": self.task_name,
"inputs": self.inputs,
"outputs": self.outputs,
"memory": self.memory,
"status": self.status,
"error": self.error,
"workspace": self.workspace,
}
# =========================================================
# Context Factory
# =========================================================
def create_context(task_name: str, inputs: Dict[str, Any]) -> ExecutionContext:
"""
Standardized factory used by executor.
"""
return ExecutionContext(
task_name=task_name,
inputs=inputs,
) |