Skip to content

store

numerous.tasks.store

Task store implementations for task and instance storage.

InMemoryTaskStore

In-memory task store for local execution.

Source code in numerous/tasks/store.py 
 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
class InMemoryTaskStore:
    """In-memory task store for local execution."""

    def __init__(self) -> None:
        self._task_definitions: dict[str, TaskDefinition] = {}
        self._task_instances: dict[str, TaskInstanceState] = {}
        self.lock = threading.Lock()

    def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
        """Register a task definition."""
        with self.lock:
            self._task_definitions[task_def.id] = task_def
        return task_def

    def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
        """Get a task definition by ID."""
        with self.lock:
            return self._task_definitions.get(task_id)

    def list_task_definitions(self) -> list[TaskDefinition]:
        """List all task definitions."""
        with self.lock:
            return list(self._task_definitions.values())

    def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
        """Register a task instance."""
        with self.lock:
            self._task_instances[state.id] = state
        return state

    def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
        """Get a task instance by ID."""
        with self.lock:
            return self._task_instances.get(instance_id)

    def list_task_instances(
        self, task_id: Optional[str] = None
    ) -> list[TaskInstanceState]:
        """List task instances, optionally filtered by task_id."""
        with self.lock:
            if task_id:
                return [
                    inst
                    for inst in self._task_instances.values()
                    if inst.task_id == task_id
                ]
            return list(self._task_instances.values())

get_task_definition(task_id)

Get a task definition by ID.

Source code in numerous/tasks/store.py 
71
72
73
74
def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
    """Get a task definition by ID."""
    with self.lock:
        return self._task_definitions.get(task_id)

get_task_instance(instance_id)

Get a task instance by ID.

Source code in numerous/tasks/store.py 
87
88
89
90
def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
    """Get a task instance by ID."""
    with self.lock:
        return self._task_instances.get(instance_id)

list_task_definitions()

List all task definitions.

Source code in numerous/tasks/store.py 
76
77
78
79
def list_task_definitions(self) -> list[TaskDefinition]:
    """List all task definitions."""
    with self.lock:
        return list(self._task_definitions.values())

list_task_instances(task_id=None)

List task instances, optionally filtered by task_id.

Source code in numerous/tasks/store.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
def list_task_instances(
    self, task_id: Optional[str] = None
) -> list[TaskInstanceState]:
    """List task instances, optionally filtered by task_id."""
    with self.lock:
        if task_id:
            return [
                inst
                for inst in self._task_instances.values()
                if inst.task_id == task_id
            ]
        return list(self._task_instances.values())

register_instance(state)

Register a task instance.

Source code in numerous/tasks/store.py 
81
82
83
84
85
def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
    """Register a task instance."""
    with self.lock:
        self._task_instances[state.id] = state
    return state

register_task(task_def)

Register a task definition.

Source code in numerous/tasks/store.py 
65
66
67
68
69
def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
    """Register a task definition."""
    with self.lock:
        self._task_definitions[task_def.id] = task_def
    return task_def

PlatformTaskStore

Platform task store that queries backend for all data.

This store does not allow local registration of tasks or instances, as all data comes from the backend.

Source code in numerous/tasks/store.py 
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
class PlatformTaskStore:
    """
    Platform task store that queries backend for all data.

    This store does not allow local registration of tasks or instances,
    as all data comes from the backend.
    """

    def __init__(
        self,
        client: Client,
        organization_slug: str,
        deploy_id: str,
    ) -> None:
        self._client = client
        self._org_slug = organization_slug
        self._deploy_id = deploy_id

    def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
        """
        No-op in platform mode.

        Tasks are defined during app deployment and managed by the backend.
        """
        return task_def

    def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
        """Get a task definition by ID from backend."""
        definitions = self.list_task_definitions()

        for task_def in definitions:
            if task_def.id == task_id:
                return task_def

        return None

    def list_task_definitions(self) -> list[TaskDefinition]:
        """List all task definitions from backend."""
        result = self._client.deployment_tasks(
            organization_slug=self._org_slug,
            deploy_id=self._deploy_id,
        )

        definitions = []
        for backend_task in result.tasks:
            task_def = TaskDefinition(
                id=backend_task.id,
                name=backend_task.id,
                func=_platform_task_placeholder,
                app_id=None,
                app_version_id=None,
                command=backend_task.command,
            )
            definitions.append(task_def)

        return definitions

    def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
        """
        No-op in platform mode.

        Instances are created by the platform executor via taskStart mutation.
        """
        return state

    def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
        """Get a task instance by ID from backend."""
        instance = self._client.task_instance(instance_id)
        if not instance:
            return None

        return self._backend_to_state(instance)

    def list_task_instances(
        self, task_id: Optional[str] = None
    ) -> list[TaskInstanceState]:
        """List task instances from backend."""
        if task_id is None:
            msg = (
                "Platform store requires task_id to list instances. "
                "Cannot list all instances across all tasks."
            )
            raise ValueError(msg)

        result = self._client.task_instances(
            organization_slug=self._org_slug,
            deploy_id=self._deploy_id,
            task_id=task_id,
        )

        return [self._backend_to_state(inst) for inst in result.task_instances]

    def _backend_to_state(
        self, backend_instance: TaskInstanceData
    ) -> TaskInstanceState:
        inputs = {}
        if backend_instance.input:
            inputs = deserialize_task_inputs(backend_instance.input)

        output = {}
        result = None
        if backend_instance.output:
            output = deserialize_task_output(backend_instance.output)
            result = output.get("result")

        status = self._map_status(
            backend_instance.workload.status.value,
            backend_instance.workload.exit_code,
        )

        # Parse progress (backend uses 0-100, we use 0-1)
        progress = 0.0
        if backend_instance.progress.value is not None:
            progress = backend_instance.progress.value / 100

        created_at = self._parse_created_at(backend_instance.created_at)

        state = TaskInstanceState(
            id=backend_instance.id,
            task_id=backend_instance.task.id,
            status=status,
            progress=progress,
            inputs=inputs,
            output=output,
            result=result,
            workload=TaskWorkload.REMOTE,
            created_at=created_at,
        )
        state.controller = PlatformTaskController(state, self._client)
        return state

    def _parse_created_at(self, value: Any) -> datetime:  # noqa: ANN401
        if isinstance(value, datetime):
            return value

        if isinstance(value, str):
            try:
                return datetime.fromisoformat(value.replace("Z", "+00:00"))
            except (ValueError, AttributeError):
                pass

        return datetime.now().astimezone()

    def _map_status(self, backend_status: str, exit_code: Optional[int]) -> TaskStatus:
        if exit_code is not None:
            return TaskStatus.COMPLETED if exit_code == 0 else TaskStatus.FAILED

        mapping = {
            "PENDING": TaskStatus.PENDING,
            "RUNNING": TaskStatus.RUNNING,
            "STOPPED": TaskStatus.CANCELLED,
            "ERROR": TaskStatus.FAILED,
            "UNKNOWN": TaskStatus.FAILED,
        }
        return mapping.get(backend_status, TaskStatus.FAILED)

get_task_definition(task_id)

Get a task definition by ID from backend.

Source code in numerous/tasks/store.py 
140
141
142
143
144
145
146
147
148
def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
    """Get a task definition by ID from backend."""
    definitions = self.list_task_definitions()

    for task_def in definitions:
        if task_def.id == task_id:
            return task_def

    return None

get_task_instance(instance_id)

Get a task instance by ID from backend.

Source code in numerous/tasks/store.py 
179
180
181
182
183
184
185
def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
    """Get a task instance by ID from backend."""
    instance = self._client.task_instance(instance_id)
    if not instance:
        return None

    return self._backend_to_state(instance)

list_task_definitions()

List all task definitions from backend.

Source code in numerous/tasks/store.py 
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def list_task_definitions(self) -> list[TaskDefinition]:
    """List all task definitions from backend."""
    result = self._client.deployment_tasks(
        organization_slug=self._org_slug,
        deploy_id=self._deploy_id,
    )

    definitions = []
    for backend_task in result.tasks:
        task_def = TaskDefinition(
            id=backend_task.id,
            name=backend_task.id,
            func=_platform_task_placeholder,
            app_id=None,
            app_version_id=None,
            command=backend_task.command,
        )
        definitions.append(task_def)

    return definitions

list_task_instances(task_id=None)

List task instances from backend.

Source code in numerous/tasks/store.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def list_task_instances(
    self, task_id: Optional[str] = None
) -> list[TaskInstanceState]:
    """List task instances from backend."""
    if task_id is None:
        msg = (
            "Platform store requires task_id to list instances. "
            "Cannot list all instances across all tasks."
        )
        raise ValueError(msg)

    result = self._client.task_instances(
        organization_slug=self._org_slug,
        deploy_id=self._deploy_id,
        task_id=task_id,
    )

    return [self._backend_to_state(inst) for inst in result.task_instances]

register_instance(state)

No-op in platform mode.

Instances are created by the platform executor via taskStart mutation.

Source code in numerous/tasks/store.py 
171
172
173
174
175
176
177
def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
    """
    No-op in platform mode.

    Instances are created by the platform executor via taskStart mutation.
    """
    return state

register_task(task_def)

No-op in platform mode.

Tasks are defined during app deployment and managed by the backend.

Source code in numerous/tasks/store.py 
132
133
134
135
136
137
138
def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
    """
    No-op in platform mode.

    Tasks are defined during app deployment and managed by the backend.
    """
    return task_def

TaskStore

Bases: Protocol

Protocol for task storage operations.

Source code in numerous/tasks/store.py 
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
class TaskStore(Protocol):
    """Protocol for task storage operations."""

    def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
        """Register a task definition."""
        ...

    def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
        """Get a task definition by ID."""
        ...

    def list_task_definitions(self) -> list[TaskDefinition]:
        """List all task definitions."""
        ...

    def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
        """Register a task instance."""
        ...

    def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
        """Get a task instance by ID."""
        ...

    def list_task_instances(
        self, task_id: Optional[str] = None
    ) -> list[TaskInstanceState]:
        """List task instances, optionally filtered by task_id."""
        ...

get_task_definition(task_id)

Get a task definition by ID.

Source code in numerous/tasks/store.py 
34
35
36
def get_task_definition(self, task_id: str) -> Optional[TaskDefinition]:
    """Get a task definition by ID."""
    ...

get_task_instance(instance_id)

Get a task instance by ID.

Source code in numerous/tasks/store.py 
46
47
48
def get_task_instance(self, instance_id: str) -> Optional[TaskInstanceState]:
    """Get a task instance by ID."""
    ...

list_task_definitions()

List all task definitions.

Source code in numerous/tasks/store.py 
38
39
40
def list_task_definitions(self) -> list[TaskDefinition]:
    """List all task definitions."""
    ...

list_task_instances(task_id=None)

List task instances, optionally filtered by task_id.

Source code in numerous/tasks/store.py 
50
51
52
53
54
def list_task_instances(
    self, task_id: Optional[str] = None
) -> list[TaskInstanceState]:
    """List task instances, optionally filtered by task_id."""
    ...

register_instance(state)

Register a task instance.

Source code in numerous/tasks/store.py 
42
43
44
def register_instance(self, state: TaskInstanceState) -> TaskInstanceState:
    """Register a task instance."""
    ...

register_task(task_def)

Register a task definition.

Source code in numerous/tasks/store.py 
30
31
32
def register_task(self, task_def: TaskDefinition) -> TaskDefinition:
    """Register a task definition."""
    ...