feat(V1R2C1): add orchestrator run store
This commit is contained in:
@@ -0,0 +1,32 @@
|
||||
"""V1R2C1 orchestrator run store."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from time import time
|
||||
from typing import Any, Dict
|
||||
import uuid
|
||||
|
||||
|
||||
@dataclass
|
||||
class RunRecord:
|
||||
run_id: str = field(default_factory=lambda: str(uuid.uuid4()))
|
||||
state: str = "created"
|
||||
context: Dict[str, Any] = field(default_factory=dict)
|
||||
events: list = field(default_factory=list)
|
||||
created_at: float = field(default_factory=time)
|
||||
|
||||
|
||||
class RunStore:
|
||||
def __init__(self):
|
||||
self.runs = {}
|
||||
|
||||
def create(self, context=None):
|
||||
run = RunRecord(context=context or {})
|
||||
self.runs[run.run_id] = run
|
||||
return run
|
||||
|
||||
def update_state(self, run_id, state):
|
||||
self.runs[run_id].state = state
|
||||
self.runs[run_id].events.append({"state": state, "time": time()})
|
||||
|
||||
def get(self, run_id):
|
||||
return self.runs.get(run_id)
|
||||
Reference in New Issue
Block a user