From 846ed8f037bf54e50f6397cb5061cbdaf958de6f Mon Sep 17 00:00:00 2001 From: Simon Kelly Date: Thu, 6 Aug 2026 15:25:36 +0200 Subject: [PATCH 1/5] Pin openapi-python-client below 0.29 0.29 generates `datetime.fromisoformat` calls in place of `dateutil.isoparse`. That can't parse the API's `Z`-suffixed timestamps before Python 3.11, and we support 3.10. Co-Authored-By: Claude Opus 5 (1M context) --- pyproject.toml | 4 ++- uv.lock | 83 +++++++++++--------------------------------------- 2 files changed, 20 insertions(+), 67 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 997a0be..f0fbaec 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,9 @@ dev = [ "invoke", "pytest-celery", "redis", - "openapi-python-client", + # 0.29 generates `datetime.fromisoformat` calls, which can't parse the API's + # `Z`-suffixed timestamps on Python 3.10. + "openapi-python-client<0.29", "taskbadger[cli]", "taskbadger[sentry]", ] diff --git a/uv.lock b/uv.lock index ed08f6d..2a40285 100644 --- a/uv.lock +++ b/uv.lock @@ -595,49 +595,23 @@ wheels = [ name = "openapi-python-client" version = "0.28.4" source = { registry = "https://pypi.org/simple" } -resolution-markers = [ - "python_full_version < '3.11'", -] dependencies = [ - { name = "attrs", marker = "python_full_version < '3.11'" }, - { name = "colorama", marker = "python_full_version < '3.11' and sys_platform == 'win32'" }, - { name = "httpx", marker = "python_full_version < '3.11'" }, - { name = "jinja2", marker = "python_full_version < '3.11'" }, - { name = "pydantic", marker = "python_full_version < '3.11'" }, - { name = "python-dateutil", marker = "python_full_version < '3.11'" }, - { name = "ruamel-yaml", marker = "python_full_version < '3.11'" }, - { name = "ruff", marker = "python_full_version < '3.11'" }, - { name = "shellingham", marker = "python_full_version < '3.11'" }, - { name = "typer", version = "0.25.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, + { name = "attrs" }, + { name = "colorama", marker = "sys_platform == 'win32'" }, + { name = "httpx" }, + { name = "jinja2" }, + { name = "pydantic" }, + { name = "python-dateutil" }, + { name = "ruamel-yaml" }, + { name = "ruff" }, + { name = "shellingham" }, + { name = "typer" }, ] sdist = { url = "https://files.pythonhosted.org/packages/7b/cf/c9278f87a8454cef126fb2594bc32472d681957ea03b8de68d304107a459/openapi_python_client-0.28.4.tar.gz", hash = "sha256:f78dfab5e21652806b17af07c16187b5911138fc526229126953c1b7b37d2b88", size = 125974, upload-time = "2026-05-11T19:44:08.427Z" } wheels = [ { url = "https://files.pythonhosted.org/packages/07/59/016bf13a3bfb85a0d890d492cc9fc65ccefa53ecc5eba3c9a049e5ea6317/openapi_python_client-0.28.4-py3-none-any.whl", hash = "sha256:6bb87dbb05f88e5ab5ba4e5902f0c85ea53e666d3646be49ea2bb19bc8b5bb85", size = 183212, upload-time = "2026-05-11T19:44:07.005Z" }, ] -[[package]] -name = "openapi-python-client" -version = "0.29.0" -source = { registry = "https://pypi.org/simple" } -resolution-markers = [ - "python_full_version >= '3.11'", -] -dependencies = [ - { name = "attrs", marker = "python_full_version >= '3.11'" }, - { name = "colorama", marker = "python_full_version >= '3.11' and sys_platform == 'win32'" }, - { name = "httpx", marker = "python_full_version >= '3.11'" }, - { name = "jinja2", marker = "python_full_version >= '3.11'" }, - { name = "pydantic", marker = "python_full_version >= '3.11'" }, - { name = "ruamel-yaml", marker = "python_full_version >= '3.11'" }, - { name = "ruff", marker = "python_full_version >= '3.11'" }, - { name = "shellingham", marker = "python_full_version >= '3.11'" }, - { name = "typer", version = "0.26.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/63/f6/67dfe73b5073251b6e2aba3d67191819798dd33dd09fcb2d12242662a960/openapi_python_client-0.29.0.tar.gz", hash = "sha256:4ff439a63765aaef548e69c4eedd86dfad57891dc322709450af0c2c9a72a23e", size = 125697, upload-time = "2026-05-30T20:32:10.533Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/fb/d1/7391a205622c40b8e9d1b79b892b4944d6e3637674b47a56196e2c96c5d1/openapi_python_client-0.29.0-py3-none-any.whl", hash = "sha256:1085864c1e0a42fb50e1f5eb84363b19de07ebfb8a3f82a146c8529b948b8f12", size = 182954, upload-time = "2026-05-30T20:32:09.037Z" }, -] - [[package]] name = "packaging" version = "26.2" @@ -1204,8 +1178,7 @@ celery = [ ] cli = [ { name = "rich" }, - { name = "typer", version = "0.25.1", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "typer", version = "0.26.8", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "typer" }, ] procrastinate = [ { name = "procrastinate" }, @@ -1217,8 +1190,7 @@ sentry = [ [package.dev-dependencies] dev = [ { name = "invoke" }, - { name = "openapi-python-client", version = "0.28.4", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.11'" }, - { name = "openapi-python-client", version = "0.29.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.11'" }, + { name = "openapi-python-client" }, { name = "pre-commit" }, { name = "procrastinate" }, { name = "pytest" }, @@ -1245,7 +1217,7 @@ provides-extras = ["celery", "cli", "procrastinate", "sentry"] [package.metadata.requires-dev] dev = [ { name = "invoke" }, - { name = "openapi-python-client" }, + { name = "openapi-python-client", specifier = "<0.29" }, { name = "pre-commit" }, { name = "procrastinate" }, { name = "pytest" }, @@ -1332,38 +1304,17 @@ wheels = [ name = "typer" version = "0.25.1" source = { registry = "https://pypi.org/simple" } -resolution-markers = [ - "python_full_version < '3.11'", -] dependencies = [ - { name = "annotated-doc", marker = "python_full_version < '3.11'" }, - { name = "click", marker = "python_full_version < '3.11'" }, - { name = "rich", marker = "python_full_version < '3.11'" }, - { name = "shellingham", marker = "python_full_version < '3.11'" }, + { name = "annotated-doc" }, + { name = "click" }, + { name = "rich" }, + { name = "shellingham" }, ] sdist = { url = "https://files.pythonhosted.org/packages/e4/51/9aed62104cea109b820bbd6c14245af756112017d309da813ef107d42e7e/typer-0.25.1.tar.gz", hash = "sha256:9616eb8853a09ffeabab1698952f33c6f29ffdbceb4eaeecf571880e8d7664cc", size = 122276, upload-time = "2026-04-30T19:32:16.964Z" } wheels = [ { url = "https://files.pythonhosted.org/packages/3f/f9/2b3ff4e56e5fa7debfaf9eb135d0da96f3e9a1d5b27222223c7296336e5f/typer-0.25.1-py3-none-any.whl", hash = "sha256:75caa44ed46a03fb2dab8808753ffacdbfea88495e74c85a28c5eefcf5f39c89", size = 58409, upload-time = "2026-04-30T19:32:18.271Z" }, ] -[[package]] -name = "typer" -version = "0.26.8" -source = { registry = "https://pypi.org/simple" } -resolution-markers = [ - "python_full_version >= '3.11'", -] -dependencies = [ - { name = "annotated-doc", marker = "python_full_version >= '3.11'" }, - { name = "colorama", marker = "python_full_version >= '3.11' and sys_platform == 'win32'" }, - { name = "rich", marker = "python_full_version >= '3.11'" }, - { name = "shellingham", marker = "python_full_version >= '3.11'" }, -] -sdist = { url = "https://files.pythonhosted.org/packages/7c/f7/68adc395201b20b872d68e975386832e8005ffeacedd43a1d837a32815be/typer-0.26.8.tar.gz", hash = "sha256:c244a6bd558886fe3f8780efb6bdd28bb9aff005a94eedebaa5cb32926fe2f7e", size = 202097, upload-time = "2026-06-26T09:22:45.705Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/80/87/b9fd69c92c6102a066e1b86a35243f53e70bd4c709f2a26d9f4fee4f4dc0/typer-0.26.8-py3-none-any.whl", hash = "sha256:3512ca79ac5c11113414b36e80281b872884477722440691c89d1112e321a49c", size = 122564, upload-time = "2026-06-26T09:22:44.72Z" }, -] - [[package]] name = "typing-extensions" version = "4.15.0" From 8c19ac50dd9b9baf369fea684ea8168a3a924622 Mon Sep 17 00:00:00 2001 From: Simon Kelly Date: Thu, 6 Aug 2026 15:25:44 +0200 Subject: [PATCH 2/5] Update API schema and regenerate client Adds `parent` to the task request/response models and a `parent` query filter on task list. Co-Authored-By: Claude Opus 5 (1M context) --- taskbadger.yaml | 46 +++++++++++++++++-- .../api/action_endpoints/action_cancel.py | 8 +++- .../api/action_endpoints/action_create.py | 16 +++++-- .../action_endpoints/action_partial_update.py | 16 +++++-- .../api/action_endpoints/action_update.py | 16 +++++-- .../internal/api/task_endpoints/task_list.py | 15 ++++++ .../internal/models/patched_task_request.py | 22 +++++++++ taskbadger/internal/models/task.py | 21 +++++++++ taskbadger/internal/models/task_request.py | 21 +++++++++ 9 files changed, 163 insertions(+), 18 deletions(-) diff --git a/taskbadger.yaml b/taskbadger.yaml index 750a24b..fde0ad1 100644 --- a/taskbadger.yaml +++ b/taskbadger.yaml @@ -28,6 +28,11 @@ paths: description: Number of results to return per page. schema: type: integer + - in: query + name: parent + schema: + type: string + description: Only return the tasks that are part of this task. - in: path name: project_slug schema: @@ -388,7 +393,10 @@ paths: description: '' post: operationId: action_create - description: Create an action for a task + description: '**Deprecated.** Per-job actions are being retired. Configure alerts + once at the project level with a global action instead of attaching them to + individual tasks. This endpoint still works but will be removed in a future + release.' summary: Create Action parameters: - in: path @@ -420,6 +428,7 @@ paths: security: - projectKeyAuth: [] - bearerAuth: [] + deprecated: true responses: '201': content: @@ -471,7 +480,10 @@ paths: description: '' put: operationId: action_update - description: Update an action + description: '**Deprecated.** Per-job actions are being retired. Configure alerts + once at the project level with a global action instead of attaching them to + individual tasks. This endpoint still works but will be removed in a future + release.' summary: Update Action parameters: - in: path @@ -509,6 +521,7 @@ paths: security: - projectKeyAuth: [] - bearerAuth: [] + deprecated: true responses: '200': content: @@ -518,7 +531,10 @@ paths: description: '' patch: operationId: action_partial_update - description: Update an action + description: '**Deprecated.** Per-job actions are being retired. Configure alerts + once at the project level with a global action instead of attaching them to + individual tasks. This endpoint still works but will be removed in a future + release.' summary: Update Action (partial) parameters: - in: path @@ -555,6 +571,7 @@ paths: security: - projectKeyAuth: [] - bearerAuth: [] + deprecated: true responses: '200': content: @@ -564,7 +581,10 @@ paths: description: '' delete: operationId: action_cancel - description: Cancel an action + description: '**Deprecated.** Per-job actions are being retired. Configure alerts + once at the project level with a global action instead of attaching them to + individual tasks. This endpoint still works but will be removed in a future + release.' summary: Cancel Action parameters: - in: path @@ -596,6 +616,7 @@ paths: security: - projectKeyAuth: [] - bearerAuth: [] + deprecated: true responses: '204': description: No response body @@ -685,6 +706,12 @@ components: PatchedTaskRequest: type: object properties: + parent: + type: string + minLength: 1 + nullable: true + description: ID of the task this task is part of. Tasks can only be nested + one level deep, and a task's parent can not be changed once set. name: type: string minLength: 1 @@ -791,6 +818,11 @@ components: project: type: string readOnly: true + parent: + type: string + nullable: true + description: ID of the task this task is part of. Tasks can only be nested + one level deep, and a task's parent can not be changed once set. name: type: string description: Name of the task @@ -894,6 +926,12 @@ components: TaskRequest: type: object properties: + parent: + type: string + minLength: 1 + nullable: true + description: ID of the task this task is part of. Tasks can only be nested + one level deep, and a task's parent can not be changed once set. name: type: string minLength: 1 diff --git a/taskbadger/internal/api/action_endpoints/action_cancel.py b/taskbadger/internal/api/action_endpoints/action_cancel.py index 04f62ea..6681b9b 100644 --- a/taskbadger/internal/api/action_endpoints/action_cancel.py +++ b/taskbadger/internal/api/action_endpoints/action_cancel.py @@ -59,7 +59,9 @@ def sync_detailed( ) -> Response[Any]: """Cancel Action - Cancel an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -99,7 +101,9 @@ async def asyncio_detailed( ) -> Response[Any]: """Cancel Action - Cancel an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): diff --git a/taskbadger/internal/api/action_endpoints/action_create.py b/taskbadger/internal/api/action_endpoints/action_create.py index b4ae9e7..970a537 100644 --- a/taskbadger/internal/api/action_endpoints/action_create.py +++ b/taskbadger/internal/api/action_endpoints/action_create.py @@ -68,7 +68,9 @@ def sync_detailed( ) -> Response[Action]: """Create Action - Create an action for a task + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -108,7 +110,9 @@ def sync( ) -> Action | None: """Create Action - Create an action for a task + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -143,7 +147,9 @@ async def asyncio_detailed( ) -> Response[Action]: """Create Action - Create an action for a task + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -181,7 +187,9 @@ async def asyncio( ) -> Action | None: """Create Action - Create an action for a task + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): diff --git a/taskbadger/internal/api/action_endpoints/action_partial_update.py b/taskbadger/internal/api/action_endpoints/action_partial_update.py index 3d647a1..1bf0f7a 100644 --- a/taskbadger/internal/api/action_endpoints/action_partial_update.py +++ b/taskbadger/internal/api/action_endpoints/action_partial_update.py @@ -73,7 +73,9 @@ def sync_detailed( ) -> Response[Action]: """Update Action (partial) - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -116,7 +118,9 @@ def sync( ) -> Action | None: """Update Action (partial) - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -154,7 +158,9 @@ async def asyncio_detailed( ) -> Response[Action]: """Update Action (partial) - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -195,7 +201,9 @@ async def asyncio( ) -> Action | None: """Update Action (partial) - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): diff --git a/taskbadger/internal/api/action_endpoints/action_update.py b/taskbadger/internal/api/action_endpoints/action_update.py index 3d755e6..10d90cc 100644 --- a/taskbadger/internal/api/action_endpoints/action_update.py +++ b/taskbadger/internal/api/action_endpoints/action_update.py @@ -72,7 +72,9 @@ def sync_detailed( ) -> Response[Action]: """Update Action - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -115,7 +117,9 @@ def sync( ) -> Action | None: """Update Action - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -153,7 +157,9 @@ async def asyncio_detailed( ) -> Response[Action]: """Update Action - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): @@ -194,7 +200,9 @@ async def asyncio( ) -> Action | None: """Update Action - Update an action + **Deprecated.** Per-job actions are being retired. Configure alerts once at the project level with a + global action instead of attaching them to individual tasks. This endpoint still works but will be + removed in a future release. Args: organization_slug (str): diff --git a/taskbadger/internal/api/task_endpoints/task_list.py b/taskbadger/internal/api/task_endpoints/task_list.py index a85f5c6..bf020ad 100644 --- a/taskbadger/internal/api/task_endpoints/task_list.py +++ b/taskbadger/internal/api/task_endpoints/task_list.py @@ -16,6 +16,7 @@ def _get_kwargs( *, cursor: str | Unset = UNSET, page_size: int | Unset = UNSET, + parent: str | Unset = UNSET, ) -> dict[str, Any]: params: dict[str, Any] = {} @@ -24,6 +25,8 @@ def _get_kwargs( params["page_size"] = page_size + params["parent"] = parent + params = {k: v for k, v in params.items() if v is not UNSET and v is not None} _kwargs: dict[str, Any] = { @@ -66,6 +69,7 @@ def sync_detailed( client: AuthenticatedClient, cursor: str | Unset = UNSET, page_size: int | Unset = UNSET, + parent: str | Unset = UNSET, ) -> Response[PaginatedTaskList]: """List Tasks @@ -76,6 +80,7 @@ def sync_detailed( project_slug (str): cursor (str | Unset): page_size (int | Unset): + parent (str | Unset): Raises: errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. @@ -90,6 +95,7 @@ def sync_detailed( project_slug=project_slug, cursor=cursor, page_size=page_size, + parent=parent, ) response = client.get_httpx_client().request( @@ -106,6 +112,7 @@ def sync( client: AuthenticatedClient, cursor: str | Unset = UNSET, page_size: int | Unset = UNSET, + parent: str | Unset = UNSET, ) -> PaginatedTaskList | None: """List Tasks @@ -116,6 +123,7 @@ def sync( project_slug (str): cursor (str | Unset): page_size (int | Unset): + parent (str | Unset): Raises: errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. @@ -131,6 +139,7 @@ def sync( client=client, cursor=cursor, page_size=page_size, + parent=parent, ).parsed @@ -141,6 +150,7 @@ async def asyncio_detailed( client: AuthenticatedClient, cursor: str | Unset = UNSET, page_size: int | Unset = UNSET, + parent: str | Unset = UNSET, ) -> Response[PaginatedTaskList]: """List Tasks @@ -151,6 +161,7 @@ async def asyncio_detailed( project_slug (str): cursor (str | Unset): page_size (int | Unset): + parent (str | Unset): Raises: errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. @@ -165,6 +176,7 @@ async def asyncio_detailed( project_slug=project_slug, cursor=cursor, page_size=page_size, + parent=parent, ) response = await client.get_async_httpx_client().request(**kwargs) @@ -179,6 +191,7 @@ async def asyncio( client: AuthenticatedClient, cursor: str | Unset = UNSET, page_size: int | Unset = UNSET, + parent: str | Unset = UNSET, ) -> PaginatedTaskList | None: """List Tasks @@ -189,6 +202,7 @@ async def asyncio( project_slug (str): cursor (str | Unset): page_size (int | Unset): + parent (str | Unset): Raises: errors.UnexpectedStatus: If the server returns an undocumented status code and Client.raise_on_unexpected_status is True. @@ -205,5 +219,6 @@ async def asyncio( client=client, cursor=cursor, page_size=page_size, + parent=parent, ) ).parsed diff --git a/taskbadger/internal/models/patched_task_request.py b/taskbadger/internal/models/patched_task_request.py index 40a7d72..519cf02 100644 --- a/taskbadger/internal/models/patched_task_request.py +++ b/taskbadger/internal/models/patched_task_request.py @@ -22,6 +22,8 @@ class PatchedTaskRequest: """ Attributes: + parent (None | str | Unset): ID of the task this task is part of. Tasks can only be nested one level deep, and a + task's parent can not be changed once set. name (str | Unset): Name of the task queue (str | Unset): Queue the task is from external_id (str | Unset): Identifier from the originating system (e.g. Celery task ID) for correlating with @@ -50,6 +52,7 @@ class PatchedTaskRequest: tags (PatchedTaskRequestTags | Unset): Tags for the task represented as a mapping from 'namespace' to 'value'. """ + parent: None | str | Unset = UNSET name: str | Unset = UNSET queue: str | Unset = UNSET external_id: str | Unset = UNSET @@ -68,6 +71,12 @@ class PatchedTaskRequest: def to_dict(self) -> dict[str, Any]: from ..models.patched_task_request_tags import PatchedTaskRequestTags + parent: None | str | Unset + if isinstance(self.parent, Unset): + parent = UNSET + else: + parent = self.parent + name = self.name queue = self.queue @@ -129,6 +138,8 @@ def to_dict(self) -> dict[str, Any]: field_dict: dict[str, Any] = {} field_dict.update(self.additional_properties) field_dict.update({}) + if parent is not UNSET: + field_dict["parent"] = parent if name is not UNSET: field_dict["name"] = name if queue is not UNSET: @@ -163,6 +174,16 @@ def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: from ..models.patched_task_request_tags import PatchedTaskRequestTags d = dict(src_dict) + + def _parse_parent(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + parent = _parse_parent(d.pop("parent", UNSET)) + name = d.pop("name", UNSET) queue = d.pop("queue", UNSET) @@ -258,6 +279,7 @@ def _parse_stale_timeout(data: object) -> int | None | Unset: tags = PatchedTaskRequestTags.from_dict(_tags) patched_task_request = cls( + parent=parent, name=name, queue=queue, external_id=external_id, diff --git a/taskbadger/internal/models/task.py b/taskbadger/internal/models/task.py index 718616b..67a4812 100644 --- a/taskbadger/internal/models/task.py +++ b/taskbadger/internal/models/task.py @@ -31,6 +31,8 @@ class Task: updated (datetime.datetime): url (str): public_url (str): + parent (None | str | Unset): ID of the task this task is part of. Tasks can only be nested one level deep, and a + task's parent can not be changed once set. queue (str | Unset): Queue the task is from external_id (str | Unset): Identifier from the originating system (e.g. Celery task ID) for correlating with logs @@ -67,6 +69,7 @@ class Task: updated: datetime.datetime url: str public_url: str + parent: None | str | Unset = UNSET queue: str | Unset = UNSET external_id: str | Unset = UNSET status: StatusEnum | Unset = StatusEnum.PENDING @@ -103,6 +106,12 @@ def to_dict(self) -> dict[str, Any]: public_url = self.public_url + parent: None | str | Unset + if isinstance(self.parent, Unset): + parent = UNSET + else: + parent = self.parent + queue = self.queue external_id = self.external_id @@ -174,6 +183,8 @@ def to_dict(self) -> dict[str, Any]: "public_url": public_url, } ) + if parent is not UNSET: + field_dict["parent"] = parent if queue is not UNSET: field_dict["queue"] = queue if external_id is not UNSET: @@ -229,6 +240,15 @@ def _parse_value_percent(data: object) -> int | None: public_url = d.pop("public_url") + def _parse_parent(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + parent = _parse_parent(d.pop("parent", UNSET)) + queue = d.pop("queue", UNSET) external_id = d.pop("external_id", UNSET) @@ -331,6 +351,7 @@ def _parse_stale_timeout(data: object) -> int | None | Unset: updated=updated, url=url, public_url=public_url, + parent=parent, queue=queue, external_id=external_id, status=status, diff --git a/taskbadger/internal/models/task_request.py b/taskbadger/internal/models/task_request.py index 363630c..fe22822 100644 --- a/taskbadger/internal/models/task_request.py +++ b/taskbadger/internal/models/task_request.py @@ -23,6 +23,8 @@ class TaskRequest: """ Attributes: name (str): Name of the task + parent (None | str | Unset): ID of the task this task is part of. Tasks can only be nested one level deep, and a + task's parent can not be changed once set. queue (str | Unset): Queue the task is from external_id (str | Unset): Identifier from the originating system (e.g. Celery task ID) for correlating with logs @@ -51,6 +53,7 @@ class TaskRequest: """ name: str + parent: None | str | Unset = UNSET queue: str | Unset = UNSET external_id: str | Unset = UNSET status: StatusEnum | Unset = StatusEnum.PENDING @@ -70,6 +73,12 @@ def to_dict(self) -> dict[str, Any]: name = self.name + parent: None | str | Unset + if isinstance(self.parent, Unset): + parent = UNSET + else: + parent = self.parent + queue = self.queue external_id = self.external_id @@ -133,6 +142,8 @@ def to_dict(self) -> dict[str, Any]: "name": name, } ) + if parent is not UNSET: + field_dict["parent"] = parent if queue is not UNSET: field_dict["queue"] = queue if external_id is not UNSET: @@ -167,6 +178,15 @@ def from_dict(cls: type[T], src_dict: Mapping[str, Any]) -> T: d = dict(src_dict) name = d.pop("name") + def _parse_parent(data: object) -> None | str | Unset: + if data is None: + return data + if isinstance(data, Unset): + return data + return cast(None | str | Unset, data) + + parent = _parse_parent(d.pop("parent", UNSET)) + queue = d.pop("queue", UNSET) external_id = d.pop("external_id", UNSET) @@ -261,6 +281,7 @@ def _parse_stale_timeout(data: object) -> int | None | Unset: task_request = cls( name=name, + parent=parent, queue=queue, external_id=external_id, status=status, From d6dd2b730470f6d118d2126b7d65c8ae2d97fde3 Mon Sep 17 00:00:00 2001 From: Simon Kelly Date: Thu, 6 Aug 2026 15:25:54 +0200 Subject: [PATCH 3/5] Add `parent` to task create, update and list Also exports `list_tasks`, which the new parent filter makes the natural way to fetch a task's children. Co-Authored-By: Claude Opus 5 (1M context) --- integration_tests/test_parents.py | 45 +++++++++++++++++++++++++++++++ taskbadger/__init__.py | 3 ++- taskbadger/sdk.py | 27 ++++++++++++++++--- 3 files changed, 71 insertions(+), 4 deletions(-) create mode 100644 integration_tests/test_parents.py diff --git a/integration_tests/test_parents.py b/integration_tests/test_parents.py new file mode 100644 index 0000000..887108d --- /dev/null +++ b/integration_tests/test_parents.py @@ -0,0 +1,45 @@ +import pytest + +import taskbadger as badger +from taskbadger.exceptions import UnexpectedStatus + + +def test_create_child_task(): + parent = badger.create_task("test parent") + child = badger.create_task("test child", parent=parent.id) + + assert child.parent == parent.id + assert badger.get_task(child.id).parent == parent.id + # the parent itself stays a root + assert not badger.get_task(parent.id).parent + + +def test_list_tasks_by_parent(): + parent = badger.create_task("test parent for listing") + child = badger.create_task("test child for listing", parent=parent.id) + badger.create_task("test unrelated task") + + children = badger.list_tasks(parent=parent.id).results + + assert [task.id for task in children] == [child.id] + + +def test_set_parent_on_an_existing_task(): + parent = badger.create_task("test parent for update") + child = badger.create_task("test child for update") + assert not child.parent + + child.update(parent=parent.id) + + assert child.parent == parent.id + assert badger.get_task(child.id).parent == parent.id + + +def test_nesting_is_limited_to_one_level(): + """The API rejects a grandchild, which is what the integrations' flattening + exists to avoid.""" + parent = badger.create_task("test parent depth") + child = badger.create_task("test child depth", parent=parent.id) + + with pytest.raises(UnexpectedStatus): + badger.create_task("test grandchild depth", parent=child.id) diff --git a/taskbadger/__init__.py b/taskbadger/__init__.py index 9f541f3..102b750 100644 --- a/taskbadger/__init__.py +++ b/taskbadger/__init__.py @@ -3,7 +3,7 @@ from .internal.models import StatusEnum from .mug import Badger, Session from .safe_sdk import create_task_safe, update_task_safe -from .sdk import DefaultMergeStrategy, Task, create_task, get_task, init, update_task +from .sdk import DefaultMergeStrategy, Task, create_task, get_task, init, list_tasks, update_task __all__ = [ "track", @@ -20,6 +20,7 @@ "create_task", "get_task", "init", + "list_tasks", "update_task", ] diff --git a/taskbadger/sdk.py b/taskbadger/sdk.py index 12c8fe6..a184440 100644 --- a/taskbadger/sdk.py +++ b/taskbadger/sdk.py @@ -162,6 +162,7 @@ def create_task( tags: dict[str, str] = None, queue: str = None, external_id: str = None, + parent: str = None, ) -> "Task": """Create a Task. @@ -178,6 +179,9 @@ def create_task( tags: Dictionary of namespace -> value tags. queue: Name of the queue the task is from. external_id: Identifier from the originating system (e.g. Celery task ID) for correlating with logs. + parent: ID of the parent task. Tasks nest a single level deep, so this must be the ID + of a task that is not itself a child. The Celery and Procrastinate integrations set + this automatically for tasks enqueued from within a tracked task. Returns: Task: The created Task object. @@ -186,6 +190,8 @@ def create_task( "name": name, "status": status, } + if parent is not None: + task_dict["parent"] = parent if queue is not None: task_dict["queue"] = queue if external_id is not None: @@ -236,6 +242,7 @@ def update_task( tags: dict[str, str] = None, queue: str = None, external_id: str = None, + parent: str = None, ) -> "Task": """Update a task. Requires only the task ID and fields to update. @@ -253,6 +260,8 @@ def update_task( tags: Dictionary of namespace -> value tags. queue: Name of the queue the task is from. external_id: Identifier from the originating system (e.g. Celery task ID) for correlating with logs. + parent: ID of the parent task. Can only be set on a task that doesn't already have a + parent — the API rejects an attempt to change one. Returns: Task: The updated Task object. @@ -266,6 +275,7 @@ def update_task( stale_timeout = _none_to_unset(stale_timeout) queue = _none_to_unset(queue) external_id = _none_to_unset(external_id) + parent = _none_to_unset(parent) data = data or UNSET body = PatchedTaskRequest( @@ -278,6 +288,7 @@ def update_task( stale_timeout=stale_timeout, queue=queue, external_id=external_id, + parent=parent, ) if actions: _warn_actions_deprecated() @@ -291,9 +302,15 @@ def update_task( return Task(response.parsed) -def list_tasks(page_size: int = None, cursor: str = None): - """List tasks.""" - kwargs = _make_args(page_size=page_size, cursor=cursor) +def list_tasks(page_size: int = None, cursor: str = None, parent: str = None): + """List tasks. + + Arguments: + page_size: Number of results to return per page. + cursor: Pagination cursor. + parent: Only return the children of this task. + """ + kwargs = _make_args(page_size=page_size, cursor=cursor, parent=parent) with Session() as client: response = task_list.sync_detailed(client=client, **kwargs) _check_response(response) @@ -353,6 +370,7 @@ def create( tags: dict[str, str] = None, queue: str = None, external_id: str = None, + parent: str = None, ) -> "Task": """Create a new task @@ -371,6 +389,7 @@ def create( tags=tags, queue=queue, external_id=external_id, + parent=parent, ) def __init__(self, task): @@ -468,6 +487,7 @@ def update( tags: dict[str, str] = None, queue: str = None, external_id: str = None, + parent: str = None, data_merge_strategy: Any = None, ): """Generic update method used to update any of the task fields. @@ -497,6 +517,7 @@ def update( tags=tags, queue=queue, external_id=external_id, + parent=parent, ) self._task = task._task From d1c59ddb16e1cc4e8777a9e8184f4ac257b3d104 Mon Sep 17 00:00:00 2001 From: Simon Kelly Date: Thu, 6 Aug 2026 15:26:38 +0200 Subject: [PATCH 4/5] Nest tasks enqueued from inside a tracked task Celery and Procrastinate tasks enqueued while another tracked task is running are attached to it as children, as are tasks created inside `@track`. Nesting is one level deep, so entering a task that already has a parent keeps offering that parent: a task enqueued by a child joins it under the root rather than hanging off it, which the API would reject. Replaces Procrastinate's private current-task contextvar with the shared one so `current_task()` and parenting can't drift apart. Co-Authored-By: Claude Opus 5 (1M context) --- integration_tests/tasks.py | 17 ++ integration_tests/test_celery.py | 26 ++- taskbadger/_current_task.py | 53 +++++ taskbadger/_integrations.py | 11 + taskbadger/celery.py | 33 ++- taskbadger/decorators.py | 24 ++- taskbadger/procrastinate.py | 35 +++- tests/test_parents.py | 343 +++++++++++++++++++++++++++++++ 8 files changed, 521 insertions(+), 21 deletions(-) create mode 100644 taskbadger/_current_task.py create mode 100644 tests/test_parents.py diff --git a/integration_tests/tasks.py b/integration_tests/tasks.py index e708cd4..fff613a 100644 --- a/integration_tests/tasks.py +++ b/integration_tests/tasks.py @@ -22,6 +22,23 @@ def add_auto_track(self, x, y): return x + y +@shared_task(bind=True, base=taskbadger.celery.Task) +def spawns_grandchild(self, x, y): + """Deferred by `spawns_child`, and defers a task of its own in turn. + + Since this is already a child, the task it defers has to be flattened onto + the root rather than nested under this one — the API only allows one level. + """ + grandchild = add.delay(x, y) + return {"own_tb_id": self.taskbadger_task_id, "grandchild_tb_id": grandchild.taskbadger_task_id} + + +@shared_task(bind=True, base=taskbadger.celery.Task) +def spawns_child(self, x, y): + """Defers a task from inside its own run, so that task nests under this one.""" + return spawns_grandchild.delay(x, y).id + + @shared_task(bind=True, base=taskbadger.celery.Task, taskbadger_heartbeat_interval=HEARTBEAT_INTERVAL) def slow_add(self, x, y): """Runs long enough to go stale without a heartbeat, and never updates itself.""" diff --git a/integration_tests/test_celery.py b/integration_tests/test_celery.py index c23a276..13d8f45 100644 --- a/integration_tests/test_celery.py +++ b/integration_tests/test_celery.py @@ -7,7 +7,7 @@ import taskbadger from taskbadger import StatusEnum -from .tasks import HEARTBEAT_INTERVAL, add, add_auto_track, slow_add +from .tasks import HEARTBEAT_INTERVAL, add, add_auto_track, slow_add, spawns_child @pytest.fixture(autouse=True) @@ -45,6 +45,30 @@ def test_celery_auto_track(celery_session_app, celery_session_worker): assert result.get(timeout=10, propagate=True) == a + b +def test_celery_child_task_nests_under_its_parent(celery_session_app, celery_session_worker): + a, b = random.randint(1, 1000), random.randint(1, 1000) + root = spawns_child.delay(a, b) + child_celery_id = root.get(timeout=15, propagate=True) + + child = celery_session_app.AsyncResult(child_celery_id).get(timeout=15, propagate=True) + + assert taskbadger.get_task(child["own_tb_id"]).parent == root.taskbadger_task_id + + +def test_celery_grandchild_is_flattened_onto_the_root(celery_session_app, celery_session_worker): + """Nesting stops at one level, so a task deferred by a child joins it under + the root instead of hanging off it (which the API would reject).""" + a, b = random.randint(1, 1000), random.randint(1, 1000) + root = spawns_child.delay(a, b) + child_celery_id = root.get(timeout=15, propagate=True) + + child = celery_session_app.AsyncResult(child_celery_id).get(timeout=15, propagate=True) + grandchild = taskbadger.get_task(child["grandchild_tb_id"]) + + assert grandchild.parent == root.taskbadger_task_id + assert grandchild.parent != child["own_tb_id"] + + def test_celery_heartbeat(celery_session_app, celery_session_worker): """The worker pings the task while it runs, so it doesn't go stale.""" a, b = random.randint(1, 1000), random.randint(1, 1000) diff --git a/taskbadger/_current_task.py b/taskbadger/_current_task.py new file mode 100644 index 0000000..3613bc9 --- /dev/null +++ b/taskbadger/_current_task.py @@ -0,0 +1,53 @@ +"""Tracks which tracked task is running in the current context so that tasks +created while it runs can be attached to it as children. Not part of the public +API. + +Task Badger nests tasks a single level deep, so `parent_id` never returns the +id of a task that is itself a child: entering a task that already has a parent +keeps that parent as the id offered for nesting. A task deferred by a child +therefore lands alongside it under the same root rather than being rejected. + +Call sites resolve a task's own parent themselves — they all have it to hand +already (from the create response, or from the cache the status update fills), +which keeps this module free of API calls. +""" + +from contextvars import ContextVar + +# (id of the running task, id of its parent or None if it is a root task) +_current: ContextVar[tuple[str, str | None] | None] = ContextVar("taskbadger_current_task", default=None) + + +def enter_task(task_id: str, parent: str = None): + """Mark `task_id` as the task running in this context. + + Arguments: + task_id: The running task. + parent: The running task's own parent, if it has one. + + Returns: + A token to pass to `exit_task`. + """ + return _current.set((task_id, parent)) + + +def exit_task(token) -> None: + _current.reset(token) + + +def current_task_id() -> str | None: + """The id of the tracked task running in this context, if any.""" + current = _current.get() + return current[0] if current else None + + +def parent_id() -> str | None: + """The id a task created right now should use as its `parent`. + + `None` outside a tracked task. + """ + current = _current.get() + if current is None: + return None + task_id, parent = current + return parent or task_id diff --git a/taskbadger/_integrations.py b/taskbadger/_integrations.py index 400ca29..c9d52ff 100644 --- a/taskbadger/_integrations.py +++ b/taskbadger/_integrations.py @@ -77,6 +77,17 @@ def safe_get_task(task_id: str): return task +def parent_of(task) -> str | None: + """The id of ``task``'s parent, or ``None`` if it has none. + + Accepts ``None`` (e.g. a failed fetch) and normalizes the generated model's + ``UNSET`` — returned for tasks fetched before the API grew the field — to + ``None``. + """ + parent = getattr(task, "parent", None) if task is not None else None + return parent or None + + def match_task_name(task_name: str, includes, excludes) -> bool: """Return True if ``task_name`` should be tracked under the given rules. diff --git a/taskbadger/celery.py b/taskbadger/celery.py index a0079bd..062d9fa 100644 --- a/taskbadger/celery.py +++ b/taskbadger/celery.py @@ -14,9 +14,10 @@ from kombu import serialization from . import sdk +from ._current_task import enter_task, exit_task, parent_id from ._error_context import capture_error_data, reset_error_context, start_error_context from ._heartbeat import heartbeat -from ._integrations import TERMINAL_STATES, resolve_heartbeat_options, safe_get_task, task_cache +from ._integrations import TERMINAL_STATES, parent_of, resolve_heartbeat_options, safe_get_task, task_cache from .internal.models import StatusEnum from .mug import Badger from .safe_sdk import create_task_safe, update_task_safe @@ -34,6 +35,9 @@ # Token returned by start_error_context(), stashed on the request so # task_postrun_handler can restore the previous context (see comment there). TB_ERROR_CTX_TOKEN = f"{KWARG_PREFIX}error_ctx_token" +# Token returned by enter_task(), stashed on the request so task_postrun_handler +# can clear the task again once it has finished running. +TB_CURRENT_TASK_TOKEN = f"{KWARG_PREFIX}current_task_token" log = logging.getLogger("taskbadger") @@ -167,6 +171,11 @@ def task_publish_handler(sender=None, headers=None, body=None, **kwargs): if routing_key and "queue" not in kwargs: kwargs["queue"] = routing_key kwargs.setdefault("external_id", headers["id"]) + # `before_task_publish` fires in the process doing the publishing, so if that + # is itself a tracked task this nests the new task under it. + enclosing_task = parent_id() + if enclosing_task: + kwargs.setdefault("parent", enclosing_task) name = kwargs.pop("name", headers["task"]) # `heartbeat_interval` isn't a task field; it tells the worker how often to @@ -280,7 +289,15 @@ def _maybe_create_task(signal_sender): delivery_info = getattr(signal_sender.request, "delivery_info", None) or {} queue = delivery_info.get("routing_key") external_id = signal_sender.request.id - create_kwargs = {"status": StatusEnum.PENDING, "data": data, "queue": queue, "external_id": external_id} + create_kwargs = { + "status": StatusEnum.PENDING, + "data": data, + "queue": queue, + "external_id": external_id, + # eager and canvas tasks are created here rather than at publish time, but + # still run inside whatever task invoked them + "parent": parent_id(), + } # `before_task_publish` never ran for these, so per-call options are still # sitting in the headers rather than resolved into the message. header_kwargs = headers.get(TB_KWARGS_ARG) or {} @@ -305,7 +322,8 @@ def task_prerun_handler(sender=None, **kwargs): _maybe_create_task(sender) _update_task(sender, StatusEnum.PROCESSING) _start_heartbeat(sender) - if _get_taskbadger_task_id(sender.request): + task_id = _get_taskbadger_task_id(sender.request) + if task_id: # Snapshotted here (same thread as the task body and the failure/retry # signals below) so context providers can tell a fresh capture from a # stale one if the task errors. The token is restored in @@ -315,6 +333,11 @@ def task_prerun_handler(sender=None, **kwargs): # clobbered by the inner task's snapshot for the rest of its run. token = start_error_context() sender.request.update({TB_ERROR_CTX_TOKEN: token}) + # Mark this as the running task so anything it publishes is nested under + # it. `_update_task` above has already put it in the cache, so this is a + # lookup rather than a fetch. + current_token = enter_task(task_id, parent_of(safe_get_task(task_id))) + sender.request.update({TB_CURRENT_TASK_TOKEN: current_token}) @task_postrun.connect @@ -323,6 +346,10 @@ def task_postrun_handler(sender=None, **kwargs): task_id = _get_taskbadger_task_id(sender.request) if task_id: heartbeat.stop(task_id) + current_token = sender.request.get(TB_CURRENT_TASK_TOKEN) + if current_token is not None: + exit_task(current_token) + sender.request.update({TB_CURRENT_TASK_TOKEN: None}) token = sender.request.get(TB_ERROR_CTX_TOKEN) if token is not None: reset_error_context(token) diff --git a/taskbadger/decorators.py b/taskbadger/decorators.py index 49174c4..ef3b155 100644 --- a/taskbadger/decorators.py +++ b/taskbadger/decorators.py @@ -1,7 +1,9 @@ import logging from functools import wraps +from ._current_task import enter_task, exit_task, parent_id from ._error_context import capture_error_data, reset_error_context, start_error_context +from ._integrations import parent_of from .mug import Session from .safe_sdk import create_task_safe from .sdk import StatusEnum @@ -46,14 +48,20 @@ def _decorator(func): @wraps(func) @Session() def _inner(*args, **kwargs): - task = create_task_safe( - task_name, - status=StatusEnum.PROCESSING, - max_runtime=max_runtime, - monitor_id=monitor_id, - **task_kwargs, - ) + create_kwargs = { + "status": StatusEnum.PROCESSING, + "max_runtime": max_runtime, + "monitor_id": monitor_id, + } + enclosing_task = parent_id() + if enclosing_task: + create_kwargs["parent"] = enclosing_task + # an explicit `parent` in task_kwargs wins over the enclosing task + create_kwargs.update(task_kwargs) + task = create_task_safe(task_name, **create_kwargs) token = start_error_context() + # so that tasks created while `func` runs are nested under this one + task_token = enter_task(task.id, parent_of(task)) if task else None try: result = func(*args, **kwargs) except Exception as e: @@ -65,6 +73,8 @@ def _inner(*args, **kwargs): ) raise finally: + if task_token is not None: + exit_task(task_token) reset_error_context(token) _update_task(task, status=StatusEnum.SUCCESS) diff --git a/taskbadger/procrastinate.py b/taskbadger/procrastinate.py index 3ce993b..6479458 100644 --- a/taskbadger/procrastinate.py +++ b/taskbadger/procrastinate.py @@ -15,12 +15,13 @@ import inspect import json import logging -from contextvars import ContextVar +from ._current_task import current_task_id, enter_task, exit_task, parent_id from ._error_context import capture_error_data, reset_error_context, start_error_context from ._heartbeat import heartbeat from ._integrations import ( TERMINAL_STATES, + parent_of, resolve_heartbeat_options, safe_get_task, task_cache, @@ -45,8 +46,6 @@ _OPTS_ATTR = "_taskbadger_opts" _SYSTEM_ATTR = "_taskbadger_system" -_current_tb_task_id: ContextVar[str | None] = ContextVar("_current_tb_task_id", default=None) - def _instrument_task(task, system=None, manual=False, opts=None): """Wrap a Procrastinate Task's ``func`` so the worker side updates TaskBadger. @@ -85,9 +84,9 @@ async def wrapped(*args, **kwargs): tb_id = kwargs.pop(TB_TASK_ID_KWARG, None) if tb_id is None: return await original_func(*args, **kwargs) - token = _current_tb_task_id.set(tb_id) + _update_status(tb_id, StatusEnum.PROCESSING) + token = _enter_running_task(tb_id) try: - _update_status(tb_id, StatusEnum.PROCESSING) heartbeat.start(tb_id, _heartbeat_interval(task)) ctx_token = start_error_context() try: @@ -101,7 +100,7 @@ async def wrapped(*args, **kwargs): return result finally: heartbeat.stop(tb_id) - _current_tb_task_id.reset(token) + exit_task(token) else: @functools.wraps(original_func) @@ -109,9 +108,9 @@ def wrapped(*args, **kwargs): tb_id = kwargs.pop(TB_TASK_ID_KWARG, None) if tb_id is None: return original_func(*args, **kwargs) - token = _current_tb_task_id.set(tb_id) + _update_status(tb_id, StatusEnum.PROCESSING) + token = _enter_running_task(tb_id) try: - _update_status(tb_id, StatusEnum.PROCESSING) heartbeat.start(tb_id, _heartbeat_interval(task)) ctx_token = start_error_context() try: @@ -125,13 +124,25 @@ def wrapped(*args, **kwargs): return result finally: heartbeat.stop(tb_id) - _current_tb_task_id.reset(token) + exit_task(token) _wrap_defer(task) task.func = wrapped setattr(task, _INSTRUMENTED_ATTR, True) +def _enter_running_task(tb_id): + """Mark ``tb_id`` as the task running in this context so anything it defers + nests under it. + + Call this after the PROCESSING update, which leaves the task in the cache, + so resolving its parent doesn't cost a fetch. Skipped entirely when Badger + isn't configured — there is nothing to fetch with. + """ + parent = parent_of(safe_get_task(tb_id)) if Badger.is_configured() else None + return enter_task(tb_id, parent) + + def _update_status(tb_id, status, exception=None): """Update the TaskBadger task to ``status``. Skips if already terminal.""" if not Badger.is_configured(): @@ -217,6 +228,10 @@ def _create_pending_task(task, task_kwargs, queue=None): opts = dict(getattr(task, _OPTS_ATTR, {}) or {}) name = opts.pop("name", None) or task.name create_kwargs = {"status": StatusEnum.PENDING} + # a defer from inside a running task nests the deferred task under it + enclosing_task = parent_id() + if enclosing_task: + create_kwargs["parent"] = enclosing_task queue = queue or getattr(task, "queue", None) if queue is not None: create_kwargs["queue"] = queue @@ -331,7 +346,7 @@ def current_task(): Returns ``None`` outside of a tracked task or if the task can't be fetched. Result is cached for the lifetime of the worker process via an LRU. """ - tb_id = _current_tb_task_id.get() + tb_id = current_task_id() if tb_id is None: return None return safe_get_task(tb_id) diff --git a/tests/test_parents.py b/tests/test_parents.py new file mode 100644 index 0000000..10729c0 --- /dev/null +++ b/tests/test_parents.py @@ -0,0 +1,343 @@ +"""Tests for parent/child task nesting. + +Tasks nest a single level deep. A task created while another one is running is +attached to it via `parent`; if the running task is itself a child, the new task +is attached to the same root rather than to the child. +""" + +import logging +from unittest import mock + +import procrastinate +import pytest +from procrastinate import testing + +# imported as a module, not by name: `test_celery_system_integration` drops +# `taskbadger.celery` from `sys.modules` and lets it be re-imported, so anything +# bound at import time can end up referring to a stale module object that +# `mock.patch` no longer reaches +import taskbadger.celery +from taskbadger import StatusEnum, track +from taskbadger._current_task import current_task_id, enter_task, exit_task, parent_id +from taskbadger.procrastinate import _instrument_task +from taskbadger.sdk import Task, create_task, list_tasks, update_task +from tests.utils import task_for_test + + +@pytest.fixture(autouse=True) +def _check_log_errors(caplog): + yield + errors = [r.getMessage() for r in caplog.get_records("call") if r.levelno == logging.ERROR] + if errors: + pytest.fail(f"log errors during tests: {errors}") + + +@pytest.fixture +def app(): + app = procrastinate.App(connector=testing.InMemoryConnector()) + with app.open(): + yield app + + +def _json_task_response(**kwargs): + response = { + "id": "test_id", + "organization": "org", + "project": "proj", + "name": "demo task", + "status": "pending", + "value": None, + "value_max": 100, + "value_percent": None, + "data": None, + "created": "2022-09-22T06:53:40.683555Z", + "updated": "2022-09-22T06:53:40.683555Z", + "url": None, + "public_url": None, + "tags": {}, + } + response.update(kwargs) + return response + + +# --- the ambient current-task record ----------------------------------------- + + +def test_no_parent_outside_a_tracked_task(): + assert current_task_id() is None + assert parent_id() is None + + +def test_tasks_created_while_a_root_task_runs_nest_under_it(): + token = enter_task("root") + try: + assert current_task_id() == "root" + assert parent_id() == "root" + finally: + exit_task(token) + assert current_task_id() is None + assert parent_id() is None + + +def test_grandchildren_are_flattened_onto_the_root(): + root = enter_task("root") + child = enter_task("child", parent="root") + try: + # the child is what's running, but nesting stops at one level so anything + # it creates joins it under the root instead of hanging off it + assert current_task_id() == "child" + assert parent_id() == "root" + finally: + exit_task(child) + exit_task(root) + + +# --- the SDK surface ---------------------------------------------------------- + + +@pytest.mark.usefixtures("_bind_settings") +def test_create_task_sends_parent(httpx_mock): + httpx_mock.add_response( + url="https://taskbadger.net/api/org/proj/tasks/", + method="POST", + match_json={"name": "child", "status": "pending", "parent": "parent_id"}, + json=_json_task_response(parent="parent_id"), + status_code=201, + ) + task = create_task("child", parent="parent_id") + assert task.parent == "parent_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_task_create_sends_parent(httpx_mock): + httpx_mock.add_response( + url="https://taskbadger.net/api/org/proj/tasks/", + method="POST", + match_json={"name": "child", "status": "pending", "parent": "parent_id"}, + json=_json_task_response(parent="parent_id"), + status_code=201, + ) + assert Task.create("child", parent="parent_id").parent == "parent_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_update_task_sends_parent(httpx_mock): + """The API allows setting a parent on a task that doesn't have one yet.""" + httpx_mock.add_response( + url="https://taskbadger.net/api/org/proj/tasks/test_id/", + method="PATCH", + match_json={"parent": "parent_id"}, + json=_json_task_response(parent="parent_id"), + status_code=200, + ) + assert update_task("test_id", parent="parent_id").parent == "parent_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_list_tasks_filters_by_parent(httpx_mock): + httpx_mock.add_response( + url="https://taskbadger.net/api/org/proj/tasks/?parent=parent_id", + method="GET", + json={"next": None, "previous": None, "results": [_json_task_response(parent="parent_id")]}, + status_code=200, + ) + (child,) = list_tasks(parent="parent_id").results + assert child.parent == "parent_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_create_task_omits_parent_when_not_given(httpx_mock): + """A bare `create_task` stays top-level even inside a running task — only the + integrations nest automatically.""" + httpx_mock.add_response( + url="https://taskbadger.net/api/org/proj/tasks/", + method="POST", + match_json={"name": "solo", "status": "pending"}, + json=_json_task_response(), + status_code=201, + ) + token = enter_task("root") + try: + create_task("solo") + finally: + exit_task(token) + + +# --- the @track decorator ----------------------------------------------------- + + +@mock.patch("taskbadger.decorators._update_safe") +@mock.patch("taskbadger.decorators.create_task_safe") +def test_track_makes_its_task_current(create, update): + create.return_value = task_for_test(id="outer_id") + seen = {} + + @track + def outer(): + seen["parent"] = parent_id() + + outer() + + assert "parent" not in create.call_args.kwargs + assert seen["parent"] == "outer_id" + + +@mock.patch("taskbadger.decorators._update_safe") +@mock.patch("taskbadger.decorators.create_task_safe") +def test_track_nests_under_an_enclosing_tracked_task(create, update): + create.side_effect = [task_for_test(id="outer_id"), task_for_test(id="inner_id", parent="outer_id")] + + @track + def inner(): + pass + + @track + def outer(): + inner() + + outer() + + assert create.call_args_list[1].kwargs["parent"] == "outer_id" + + +@mock.patch("taskbadger.decorators._update_safe") +@mock.patch("taskbadger.decorators.create_task_safe") +def test_track_explicit_parent_wins(create, update): + create.return_value = task_for_test(id="outer_id") + + @track(parent="chosen") + def inner(): + pass + + token = enter_task("root") + try: + inner() + finally: + exit_task(token) + + assert create.call_args.kwargs["parent"] == "chosen" + + +# --- Celery ------------------------------------------------------------------- + + +def _publish(name="child.task", **headers): + headers = {"id": "abc123", "task": name, "taskbadger_track": True, **headers} + taskbadger.celery.task_publish_handler(sender=name, headers=headers, body=[[], {}, {}]) + + +@pytest.mark.usefixtures("_bind_settings") +def test_celery_publish_nests_under_the_running_task(): + with mock.patch("taskbadger.celery.create_task_safe") as create: + create.return_value = task_for_test() + token = enter_task("root_id") + try: + _publish() + finally: + exit_task(token) + + assert create.call_args.kwargs["parent"] == "root_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_celery_publish_flattens_grandchildren(): + with mock.patch("taskbadger.celery.create_task_safe") as create: + create.return_value = task_for_test() + root = enter_task("root_id") + child = enter_task("child_id", parent="root_id") + try: + _publish() + finally: + exit_task(child) + exit_task(root) + + assert create.call_args.kwargs["parent"] == "root_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_celery_publish_has_no_parent_outside_a_task(): + with mock.patch("taskbadger.celery.create_task_safe") as create: + create.return_value = task_for_test() + _publish() + + assert "parent" not in create.call_args.kwargs + + +@pytest.mark.usefixtures("_bind_settings") +def test_celery_publish_explicit_parent_wins(): + with mock.patch("taskbadger.celery.create_task_safe") as create: + create.return_value = task_for_test() + token = enter_task("root_id") + try: + _publish(**{taskbadger.celery.TB_KWARGS_ARG: {"parent": "chosen"}}) + finally: + exit_task(token) + + assert create.call_args.kwargs["parent"] == "chosen" + + +# --- Procrastinate ------------------------------------------------------------ + + +@pytest.mark.usefixtures("_bind_settings") +def test_procrastinate_defer_nests_under_the_running_task(app): + @app.task(name="child") + def child(x): + return x + + _instrument_task(child, system=None, manual=True) + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe") as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + ): + create.return_value = task_for_test(id="child_tb") + token = enter_task("root_id") + try: + child.defer(x=1) + finally: + exit_task(token) + + assert create.call_args.kwargs["parent"] == "root_id" + + +@pytest.mark.usefixtures("_bind_settings") +def test_procrastinate_defer_has_no_parent_outside_a_task(app): + @app.task(name="solo") + def solo(x): + return x + + _instrument_task(solo, system=None, manual=True) + + with ( + mock.patch("taskbadger.procrastinate.create_task_safe") as create, + mock.patch("taskbadger.procrastinate.update_task_safe"), + ): + create.return_value = task_for_test(id="solo_tb") + solo.defer(x=1) + + assert "parent" not in create.call_args.kwargs + + +@pytest.mark.usefixtures("_bind_settings") +def test_procrastinate_worker_makes_its_task_current(app): + """The worker side marks the running job's task as current so anything it + defers nests under it.""" + seen = {} + + @app.task(name="records_parent") + def records_parent(): + seen["parent"] = parent_id() + + _instrument_task(records_parent, system=None, manual=True) + + with ( + mock.patch("taskbadger.procrastinate.update_task_safe") as update, + mock.patch("taskbadger.sdk.get_task") as get, + ): + update.return_value = task_for_test(id="job_tb", status=StatusEnum.PROCESSING) + get.return_value = update.return_value + records_parent.func(**{"__taskbadger_task_id__": "job_tb"}) + + assert seen["parent"] == "job_tb" + assert current_task_id() is None From 51a1166a7ecab521b6ffdbe10b4b0ec4c1d79a7f Mon Sep 17 00:00:00 2001 From: Simon Kelly Date: Thu, 6 Aug 2026 15:27:11 +0200 Subject: [PATCH 5/5] Don't nest chain links and callbacks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Celery dispatches the next chain link and any `link` callbacks from inside `trace_task` after the body returns but before `task_postrun`, so the task they follow still looks current and they were being recorded as its children. They're successors, not subtasks. Nothing on the wire distinguishes them from a publish the body made itself, so this matches against the running task's own `request.chain`/`request.callbacks` rather than inspecting the stack. Retries still nest — a retry republishes the same task, which isn't among its own successors. Co-Authored-By: Claude Opus 5 (1M context) --- integration_tests/tasks.py | 10 +++++++ integration_tests/test_celery.py | 15 ++++++++++- taskbadger/celery.py | 37 +++++++++++++++++++++++++- tests/test_parents.py | 45 ++++++++++++++++++++++++++++++++ 4 files changed, 105 insertions(+), 2 deletions(-) diff --git a/integration_tests/tasks.py b/integration_tests/tasks.py index fff613a..0c99e66 100644 --- a/integration_tests/tasks.py +++ b/integration_tests/tasks.py @@ -39,6 +39,16 @@ def spawns_child(self, x, y): return spawns_grandchild.delay(x, y).id +@shared_task(bind=True, base=taskbadger.celery.Task) +def chain_head(self): + return self.taskbadger_task_id + + +@shared_task(bind=True, base=taskbadger.celery.Task) +def chain_tail(self, head_tb_id): + return {"head_tb_id": head_tb_id, "own_tb_id": self.taskbadger_task_id} + + @shared_task(bind=True, base=taskbadger.celery.Task, taskbadger_heartbeat_interval=HEARTBEAT_INTERVAL) def slow_add(self, x, y): """Runs long enough to go stale without a heartbeat, and never updates itself.""" diff --git a/integration_tests/test_celery.py b/integration_tests/test_celery.py index 13d8f45..e1ee415 100644 --- a/integration_tests/test_celery.py +++ b/integration_tests/test_celery.py @@ -3,11 +3,12 @@ import time import pytest +from celery import chain import taskbadger from taskbadger import StatusEnum -from .tasks import HEARTBEAT_INTERVAL, add, add_auto_track, slow_add, spawns_child +from .tasks import HEARTBEAT_INTERVAL, add, add_auto_track, chain_head, chain_tail, slow_add, spawns_child @pytest.fixture(autouse=True) @@ -69,6 +70,18 @@ def test_celery_grandchild_is_flattened_onto_the_root(celery_session_app, celery assert grandchild.parent != child["own_tb_id"] +def test_celery_chain_links_are_not_nested(celery_session_app, celery_session_worker): + """Celery dispatches the next chain link from inside the previous task's run, + so it would otherwise be nested under it. Links are successors, not subtasks. + """ + ids = chain(chain_head.s(), chain_tail.s()).apply_async().get(timeout=20, propagate=True) + + assert ids["head_tb_id"], "the first link should be tracked" + assert ids["own_tb_id"], "the second link should be tracked" + assert not taskbadger.get_task(ids["own_tb_id"]).parent + assert taskbadger.list_tasks(parent=ids["head_tb_id"]).results == [] + + def test_celery_heartbeat(celery_session_app, celery_session_worker): """The worker pings the task while it runs, so it doesn't go stale.""" a, b = random.randint(1, 1000), random.randint(1, 1000) diff --git a/taskbadger/celery.py b/taskbadger/celery.py index 062d9fa..b9dee40 100644 --- a/taskbadger/celery.py +++ b/taskbadger/celery.py @@ -141,6 +141,41 @@ def taskbadger_task(self): return task +def _signature_names(signature): + """The task names ``signature`` will publish, expanding groups. + + Signatures reach us either as ``Signature`` objects or as the dicts they + serialize to, depending on where in Celery they came from. + """ + if isinstance(signature, dict): + name = signature.get("task") + nested = (signature.get("kwargs") or {}).get("tasks") + else: + name = getattr(signature, "task", None) + nested = getattr(signature, "tasks", None) + if nested: + for inner in nested: + yield from _signature_names(inner) + elif name: + yield name + + +def _is_workflow_successor(task_name): + """Whether the task being published is the running task's next chain link or + one of its callbacks, rather than something its body chose to enqueue. + + Celery dispatches both from inside ``trace_task`` after the task body + returns but before ``task_postrun``, so the running task still looks + current. They're the task's successors rather than its subtasks, so they + shouldn't be nested under it. + """ + request = getattr(celery.current_task, "request", None) + if request is None: + return False + successors = list(request.chain or []) + list(request.callbacks or []) + return any(name == task_name for sig in successors for name in _signature_names(sig)) + + @before_task_publish.connect def task_publish_handler(sender=None, headers=None, body=None, **kwargs): routing_key = kwargs.get("routing_key") @@ -174,7 +209,7 @@ def task_publish_handler(sender=None, headers=None, body=None, **kwargs): # `before_task_publish` fires in the process doing the publishing, so if that # is itself a tracked task this nests the new task under it. enclosing_task = parent_id() - if enclosing_task: + if enclosing_task and not _is_workflow_successor(sender): kwargs.setdefault("parent", enclosing_task) name = kwargs.pop("name", headers["task"]) diff --git a/tests/test_parents.py b/tests/test_parents.py index 10729c0..fbf8c58 100644 --- a/tests/test_parents.py +++ b/tests/test_parents.py @@ -263,6 +263,51 @@ def test_celery_publish_has_no_parent_outside_a_task(): assert "parent" not in create.call_args.kwargs +def _running_task(chain=None, callbacks=None): + """Pretend a task with these successors is mid-run.""" + request = mock.Mock(chain=chain, callbacks=callbacks) + return mock.patch("celery.current_task", mock.Mock(request=request)) + + +@pytest.mark.usefixtures("_bind_settings") +@pytest.mark.parametrize( + "successors", + [ + {"chain": [{"task": "child.task"}]}, + {"callbacks": [{"task": "child.task"}]}, + # a group of callbacks arrives as one signature wrapping the real tasks + {"callbacks": [{"task": "celery.group", "kwargs": {"tasks": [{"task": "child.task"}]}}]}, + ], + ids=["chain", "callback", "callback_group"], +) +def test_celery_workflow_successors_are_not_nested(successors): + """Chain links and callbacks are dispatched while the task that precedes + them still looks current, but they follow it rather than belong to it.""" + with mock.patch("taskbadger.celery.create_task_safe") as create, _running_task(**successors): + create.return_value = task_for_test() + token = enter_task("root_id") + try: + _publish() + finally: + exit_task(token) + + assert "parent" not in create.call_args.kwargs + + +@pytest.mark.usefixtures("_bind_settings") +def test_celery_publish_from_a_task_body_still_nests_while_a_chain_is_pending(): + """Only the successor itself is exempt — a task the body enqueues is not.""" + with mock.patch("taskbadger.celery.create_task_safe") as create, _running_task(chain=[{"task": "next.task"}]): + create.return_value = task_for_test() + token = enter_task("root_id") + try: + _publish() + finally: + exit_task(token) + + assert create.call_args.kwargs["parent"] == "root_id" + + @pytest.mark.usefixtures("_bind_settings") def test_celery_publish_explicit_parent_wins(): with mock.patch("taskbadger.celery.create_task_safe") as create: