Skip to content

Commit 07b6c01

Browse files
authored
feat: Add task state query API for static and dynamic tasks (#47)
* feat: Add task state query API for static and dynamic tasks Adds comprehensive query capabilities to inspect task state: - get_task(task, instance_name=None) for querying individual tasks - list_tasks(parent_task) for paginated listing of dynamic task instances - Supports both static and dynamic tasks with flexible API patterns - Implemented across all backends (PostgreSQL, MySQL, SQLite, DynamoDB, Redis) New Repository methods: - get_task_state() - query individual task state - list_task_states() - list dynamic task instances with pagination New model classes: - TaskState - snapshot of task state (task, scheduled_at, locked_until, locked_by) - ListTasksResponse - paginated results with continuation tokens Implementation highlights: - Redis uses optimized Lua script for listing with nested array returns - All backends handle static vs dynamic tasks via task_spec differentiation - Type-safe assertions ensure scheduled_at is never None for existing tasks - Comprehensive test coverage for all three usage patterns - Updated documentation with examples for static and dynamic tasks * simplify & improve coverage
1 parent a29c291 commit 07b6c01

15 files changed

Lines changed: 1054 additions & 1 deletion

‎docs/changelog.md‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,13 @@
77
- Added `force` parameter to `schedule_task` to control behavior when updating dynamic task instances
88
- Default (`force=False`): Fails with `TaskLockedException` if task is locked, preserves sooner execution time to prevent schedule postponement
99
- With `force=True`: Unconditionally overwrites task state including locks and schedule
10+
- Added new task state query API for inspecting both static and dynamic tasks
11+
- `get_task(task, instance_name=None)` - Query the state of any task (static or dynamic)
12+
- For static tasks: `get_task(my_task)`
13+
- For dynamic tasks: `get_task(template, "instance")` or `get_task(concrete_instance)`
14+
- `list_tasks(parent_task, limit=None, continuation_token=None)` - List all instances of a dynamic task with pagination
15+
- Returns `TaskState` objects containing task, scheduled_at, locked_until, and locked_by
16+
- Implemented across all backends: PostgreSQL, MySQL, SQLite, DynamoDB, Redis
1017

1118
### Bug Fixes
1219

@@ -22,6 +29,12 @@
2229

2330
- Added `TaskLockedException` exception type for clearer error handling when tasks are locked
2431
- Updated `register_task` across all backends to accept `force` parameter
32+
- Added new `get_task_state()` method to Repository interface for querying individual task state
33+
- Added new `list_task_states()` method to Repository interface for paginated listing of dynamic task instances
34+
- Added new public API methods to `PyncetteContext`:
35+
- `get_task(task, instance_name=None)` - Query state of static or dynamic tasks
36+
- `list_tasks(parent_task, limit=None, continuation_token=None)` - List dynamic task instances with pagination
37+
- Added new model classes: `TaskState` (task state snapshot) and `ListTasksResponse` (paginated results)
2538

2639
## 1.0.0 (2025-10-17)
2740

‎docs/usage.md‎

Lines changed: 112 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -411,6 +411,118 @@ With `force=True`, the task will be updated immediately even if it is executing.
411411

412412
When updating the schedule of an existing task, the sooner of the existing and new execution times is used, preventing repeated calls from indefinitely postponing execution.
413413

414+
## Querying task state
415+
416+
Pyncette provides methods to inspect the state of both static and dynamic tasks.
417+
418+
### Getting a single task
419+
420+
You can retrieve the state of any task (static or dynamic):
421+
422+
**For static tasks:**
423+
424+
```python
425+
@app.task(interval=datetime.timedelta(minutes=5))
426+
async def cleanup(context: Context) -> None:
427+
print("Running cleanup")
428+
429+
430+
async with app.create() as app_context:
431+
# Get the static task state
432+
task_state = await app_context.get_task(cleanup)
433+
434+
if task_state:
435+
print(f"Task: {task_state.task.name}")
436+
print(f"Next execution: {task_state.scheduled_at}")
437+
print(f"Locked until: {task_state.locked_until}")
438+
print(f"Locked by: {task_state.locked_by}")
439+
else:
440+
print("Task not found")
441+
```
442+
443+
**For dynamic tasks:**
444+
445+
```python
446+
@app.dynamic_task()
447+
async def hello(context: Context) -> None:
448+
print(f"Hello {context.args['username']}")
449+
450+
451+
async with app.create() as app_context:
452+
await app_context.schedule_task(
453+
hello, "user_1", interval=datetime.timedelta(minutes=5), username="Alice"
454+
)
455+
456+
# Get the dynamic task state using template + instance_name
457+
task_state = await app_context.get_task(hello, "user_1")
458+
459+
# Or using a pre-instantiated task
460+
concrete_task = hello.instantiate("user_1", interval=datetime.timedelta(minutes=5))
461+
task_state = await app_context.get_task(concrete_task)
462+
463+
if task_state:
464+
print(f"Task: {task_state.task.name}")
465+
print(f"Next execution: {task_state.scheduled_at}")
466+
print(f"Locked until: {task_state.locked_until}")
467+
print(f"Locked by: {task_state.locked_by}")
468+
print(f"Username: {task_state.task.extra_args['username']}")
469+
else:
470+
print("Task not found")
471+
```
472+
473+
The `TaskState` object contains:
474+
475+
- `task`: The task instance with all its parameters
476+
- `scheduled_at`: When the task is next scheduled to execute
477+
- `locked_until`: When the current execution lease expires (None if not locked)
478+
- `locked_by`: UUID of the worker holding the lease (None if not locked)
479+
480+
### Listing all task instances
481+
482+
You can list all instances of a dynamic task with pagination (note: this is only available for dynamic tasks, as static tasks have only a single instance):
483+
484+
```python
485+
@app.dynamic_task()
486+
async def hello(context: Context) -> None:
487+
print(f"Hello {context.args['username']}")
488+
489+
490+
async with app.create() as app_context:
491+
# Schedule multiple tasks
492+
for i in range(100):
493+
await app_context.schedule_task(
494+
hello,
495+
f"user_{i}",
496+
interval=datetime.timedelta(minutes=5),
497+
username=f"User{i}",
498+
)
499+
500+
# List all tasks with pagination
501+
response = await app_context.list_tasks(hello, limit=10)
502+
503+
for task_state in response.tasks:
504+
print(
505+
f"Task: {task_state.task.name}, Username: {task_state.task.extra_args['username']}"
506+
)
507+
508+
# Continue with next page if available
509+
while response.continuation_token:
510+
response = await app_context.list_tasks(
511+
hello, limit=10, continuation_token=response.continuation_token
512+
)
513+
for task_state in response.tasks:
514+
print(
515+
f"Task: {task_state.task.name}, Username: {task_state.task.extra_args['username']}"
516+
)
517+
```
518+
519+
Notes:
520+
521+
- Results are not guaranteed to be in any particular order
522+
- The `limit` parameter is an upper bound; backends may return fewer results
523+
- Pagination support varies by backend (DynamoDB, Redis, SQL databases all support it)
524+
- The continuation token is opaque and backend-specific
525+
414526
## Performance
415527

416528
Tasks are executed in parallel. If you have a lot of long running tasks, you can set `concurrency_limit` in `Pyncette` constructor, as this ensures that there are at most that many executing tasks at any given time. If there are no free slots in the semaphore, this will serve as a back-pressure and ensure that we don't poll additional tasks until some of the currently executing ones finish, enabling the pending tasks to be scheduled on other instances of your app. Setting `concurrency_limit` to 1 is equivalent of serializing the execution of all the tasks.

‎mkdocs.yml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
site_name: Pyncette
22
site_description: A reliable distributed scheduler with pluggable storage backends for Async Python
3-
site_url: https://pyncette.readthedocs.io
3+
site_url: https://tibordp.github.io/pyncette/
44
repo_url: https://github.com/tibordp/pyncette
55
repo_name: tibordp/pyncette
66
edit_uri: edit/master/docs/

‎src/pyncette/dynamodb.py‎

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,11 @@
2121
from pyncette.model import ContinuationToken
2222
from pyncette.model import ExecutionMode
2323
from pyncette.model import Lease
24+
from pyncette.model import ListTasksResponse
2425
from pyncette.model import PollResponse
2526
from pyncette.model import QueryResponse
2627
from pyncette.model import ResultType
28+
from pyncette.model import TaskState
2729
from pyncette.repository import Repository
2830
from pyncette.task import Task
2931

@@ -378,6 +380,97 @@ async def unlock_task(self, utc_now: datetime.datetime, task: Task, lease: Lease
378380

379381
raise PyncetteException("Unable to acquire the lock on the task due to contention")
380382

383+
async def get_task_state(
384+
self,
385+
utc_now: datetime.datetime,
386+
task: Task,
387+
) -> TaskState | None:
388+
response = await self._table.get_item(
389+
Key={
390+
"partition_id": self._get_partition_id(task),
391+
"task_id": task.canonical_name,
392+
}
393+
)
394+
395+
if "Item" not in response:
396+
return None
397+
398+
item = response["Item"]
399+
400+
# For dynamic tasks, re-instantiate from spec to ensure we have fresh parameters
401+
if task.parent_task is not None:
402+
task_spec = json.loads(item["task_spec"]) if item.get("task_spec") else None
403+
if not task_spec:
404+
raise PyncetteException(f"Task {task.canonical_name} has no task_spec stored")
405+
instantiated_task = task.parent_task.instantiate_from_spec(task_spec)
406+
else:
407+
# Static task - use as-is (no task_spec stored)
408+
instantiated_task = task
409+
410+
execute_after = item.get("execute_after")
411+
assert execute_after is not None, "execute_after should not be None for existing tasks"
412+
scheduled_at = datetime.datetime.fromisoformat(execute_after)
413+
414+
return TaskState(
415+
task=instantiated_task,
416+
scheduled_at=scheduled_at,
417+
locked_until=datetime.datetime.fromisoformat(item["locked_until"]) if item.get("locked_until") else None,
418+
locked_by=item.get("locked_by"),
419+
)
420+
421+
async def list_task_states(
422+
self,
423+
utc_now: datetime.datetime,
424+
parent_task: Task,
425+
limit: int | None = None,
426+
continuation_token: ContinuationToken | None = None,
427+
) -> ListTasksResponse:
428+
if limit is None:
429+
limit = self._batch_size
430+
431+
# Query DynamoDB on partition_id (which encodes parent_name)
432+
prefix = (self._partition_prefix or "").replace(":", "::")
433+
partition_id = f"{prefix}:{parent_task.canonical_name}"
434+
435+
query_kwargs: dict[str, Any] = {
436+
"KeyConditionExpression": Key("partition_id").eq(partition_id),
437+
"Limit": limit,
438+
}
439+
440+
if continuation_token is not None:
441+
query_kwargs["ExclusiveStartKey"] = continuation_token
442+
443+
response = await self._table.query(**query_kwargs)
444+
445+
tasks = []
446+
for item in response.get("Items", []):
447+
task_spec = json.loads(item["task_spec"]) if item.get("task_spec") else None
448+
if not task_spec:
449+
logger.warning(f"Task {item['task_id']} has no task_spec, skipping")
450+
continue
451+
452+
instantiated_task = parent_task.instantiate_from_spec(task_spec)
453+
454+
execute_after = item.get("execute_after")
455+
assert execute_after is not None, "execute_after should not be None for existing tasks"
456+
scheduled_at = datetime.datetime.fromisoformat(execute_after)
457+
458+
tasks.append(
459+
TaskState(
460+
task=instantiated_task,
461+
scheduled_at=scheduled_at,
462+
locked_until=datetime.datetime.fromisoformat(item["locked_until"]) if item.get("locked_until") else None,
463+
locked_by=item.get("locked_by"),
464+
)
465+
)
466+
467+
next_token = ContinuationToken(response["LastEvaluatedKey"]) if "LastEvaluatedKey" in response else None
468+
469+
return ListTasksResponse(
470+
tasks=tasks,
471+
continuation_token=next_token,
472+
)
473+
381474
async def extend_lease(self, utc_now: datetime.datetime, task: Task, lease: Lease) -> Lease | None:
382475
assert isinstance(lease, _TaskRecord)
383476
record: _TaskRecord | None = cast(_TaskRecord, copy.copy(lease))

‎src/pyncette/model.py‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -109,6 +109,24 @@ class QueryResponse:
109109
continuation_token: ContinuationToken | None
110110

111111

112+
@dataclass
113+
class TaskState:
114+
"""The state of a dynamic task instance"""
115+
116+
task: pyncette.task.Task
117+
scheduled_at: datetime.datetime
118+
locked_until: datetime.datetime | None
119+
locked_by: str | None
120+
121+
122+
@dataclass
123+
class ListTasksResponse:
124+
"""The result of listing dynamic task instances"""
125+
126+
tasks: list[TaskState]
127+
continuation_token: ContinuationToken | None
128+
129+
112130
if TYPE_CHECKING:
113131
import pyncette
114132
import pyncette.task

0 commit comments

Comments
 (0)