diff --git a/docs/hub/scheduler.md b/docs/hub/scheduler.md index 17abed4..5eecb74 100644 --- a/docs/hub/scheduler.md +++ b/docs/hub/scheduler.md @@ -1,14 +1,132 @@ # HUB · Scheduler -Access with `norbix.hub.scheduler`. - -| Method | Verb | Path | Scope | -| --- | --- | --- | --- | -| `disable_scheduler` | `GET` | `/{version}/scheduler/disable` | `project` | -| `enable_scheduler` | `GET` | `/{version}/scheduler/enable` | `project` | -| `delete_scheduler_task` | `DELETE` | `/{version}/scheduler/tasks/{Id}` | `project` | -| `disable_scheduler_task` | `PUT` | `/{version}/scheduler/tasks/{Id}/disable` | `project` | -| `enable_scheduler_task` | `PUT` | `/{version}/scheduler/tasks/{Id}/enable` | `project` | -| `get_scheduler_task` | `GET` | `/{version}/scheduler/tasks/{id}` | `project` | -| `get_scheduler_tasks` | `GET` | `/{version}/scheduler/tasks` | `project` | -| `save_scheduler_task` | `POST` | `/{version}/scheduler/tasks` | `project` | +Access with `norbix.hub.scheduler` (and the same methods, awaited, on +`AsyncNorbix`). The scheduler runs a task on a cron schedule. + +| Method | Verb | Path | Scope | Parameters go in | +| --- | --- | --- | --- | --- | +| `enable_scheduler` | `PUT` | `/{version}/scheduler/enable` | `project` | — | +| `disable_scheduler` | `PUT` | `/{version}/scheduler/disable` | `project` | — | +| `get_scheduler_tasks` | `GET` | `/{version}/scheduler/tasks` | `project` | query: `pageSize`, `startingAfter`, `endingBefore`, `type`, `enabled` | +| `get_scheduler_task` | `GET` | `/{version}/scheduler/tasks/{id}` | `project` | path: `id` | +| `save_scheduler_task` | `POST` | `/{version}/scheduler/tasks` | `project` | JSON body (see below) | +| `enable_scheduler_task` | `PUT` | `/{version}/scheduler/tasks/{Id}/enable` | `project` | path: `id` | +| `disable_scheduler_task` | `PUT` | `/{version}/scheduler/tasks/{Id}/disable` | `project` | path: `id` | +| `delete_scheduler_task` | `DELETE` | `/{version}/scheduler/tasks/{Id}` | `project` | path: `id` | + +Keyword arguments other than `id`, `timeout` and `bearer_token` are the +request: the query string for `GET` / `DELETE`, the JSON body otherwise. Task +ids look like `tsk_…`. Every method, its path, its query and its body are +asserted in `tests/hub/test_scheduler.py`, for the sync and the async client. + +> **Module enable / disable are `PUT`.** Versions up to 3.5 sent `GET`; the +> gateway now answers only `PUT` on `/scheduler/enable` and +> `/scheduler/disable`, so the call from an older SDK fails. Upgrade the SDK. + +## Turn the module on + +```python +norbix.hub.scheduler.enable_scheduler() +``` + +## Save a task (create or update) + +**Only `EmailCampaign` tasks run today.** `PushCampaign`, `SmsCampaign`, +`CodeFunctionalCall` and `WebhookCall` exist in the gateway's type list, but +it refuses them on save. + +The body: + +| field | required | meaning | +| --- | --- | --- | +| `initiatorUserId` | yes | the user the task runs as (`usr_…`): you, or a service user of this project | +| `name` | yes | display name | +| `cron` | yes | exactly **5 fields** (minute hour day-of-month month day-of-week), evaluated in **UTC** — `"0 9 * * 1"` is every Monday 09:00 UTC | +| `isEnabled` | yes | start enabled or not | +| `stopOnError` | yes | disable the task after a failed run | +| `task` | yes | what to run (below) | +| `taskId` | on update | the `tsk_…` id to update; leave it out to create | +| `description` | no | notes | + +`task` is a dict: + +| field | required | meaning | +| --- | --- | --- | +| `type` | yes | `"EmailCampaign"` — the name, as a string | +| `campaign` | yes | the email campaign to send: the same body the email campaign endpoints take — `source`, `templateId`, and the audience fields of that source | +| `databaseIntegrationId` | no | the database integration the campaign reads from | + +`campaign["source"]` picks the audience: `AllUsers` (optional `rolesNames`, +`userTags`), `SpecifiedUsers` (`userRecipients`), `AccountUsers` +(`userRecipients`), `Email`, `Collection`. Every source also takes the optional +`integrationId`, `language`, `notes` and `mappedTokens`. + +```python +saved = norbix.hub.scheduler.save_scheduler_task( + initiatorUserId="usr_123", + name="Weekly digest", + cron="0 9 * * 1", # Mondays 09:00 UTC + isEnabled=True, + stopOnError=False, + task={ + "type": "EmailCampaign", + "campaign": { + "source": "AllUsers", + "templateId": "etpl_123", + }, + }, +) +task_id = saved["id"] # tsk_… +``` + +This goes on the wire as: + +```json +{ + "initiatorUserId": "usr_123", + "name": "Weekly digest", + "cron": "0 9 * * 1", + "isEnabled": true, + "stopOnError": false, + "task": { + "type": "EmailCampaign", + "campaign": { "source": "AllUsers", "templateId": "etpl_123" } + } +} +``` + +To update, send the same body with `taskId="tsk_…"`. + +## Read tasks + +```python +page = norbix.hub.scheduler.get_scheduler_tasks(pageSize=20, type="EmailCampaign", enabled=True) +for row in page["list"]["items"]: + print(row["taskId"], row["name"], row["cron"], row["isEnabled"]) + +one = norbix.hub.scheduler.get_scheduler_task(id="tsk_123") +# one["item"]["payloadJson"] holds the saved task body as JSON. +``` + +## Pause, resume, delete + +```python +norbix.hub.scheduler.disable_scheduler_task(id="tsk_123") +norbix.hub.scheduler.enable_scheduler_task(id="tsk_123") +norbix.hub.scheduler.delete_scheduler_task(id="tsk_123") +``` + +## Async + +```python +async with AsyncNorbix(project_id="proj_123", api_key="sk_...") as norbix: + body = { + "initiatorUserId": "usr_123", + "name": "Weekly digest", + "cron": "0 9 * * 1", + "isEnabled": True, + "stopOnError": False, + "task": {"type": "EmailCampaign", "campaign": {"source": "AllUsers", "templateId": "etpl_123"}}, + } + saved = await norbix.hub.scheduler.save_scheduler_task(**body) +``` diff --git a/src/norbix_python/hub/scheduler.py b/src/norbix_python/hub/scheduler.py index e95d4be..6c63116 100644 --- a/src/norbix_python/hub/scheduler.py +++ b/src/norbix_python/hub/scheduler.py @@ -10,11 +10,11 @@ def __init__(self, transport: Transport) -> None: self._transport = transport def disable_scheduler(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """GET /{version}/scheduler/disable""" + """PUT /{version}/scheduler/disable""" return self._transport.send( target="hub", path="/{version}/scheduler/disable", - method="GET", + method="PUT", path_params={}, request=request, scope="project", @@ -23,11 +23,11 @@ def disable_scheduler(self, *, timeout: float | None = None, bearer_token: str | ) def enable_scheduler(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """GET /{version}/scheduler/enable""" + """PUT /{version}/scheduler/enable""" return self._transport.send( target="hub", path="/{version}/scheduler/enable", - method="GET", + method="PUT", path_params={}, request=request, scope="project", @@ -101,7 +101,17 @@ def get_scheduler_tasks(self, *, timeout: float | None = None, bearer_token: str ) def save_scheduler_task(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """POST /{version}/scheduler/tasks""" + """POST /{version}/scheduler/tasks + + Create (no ``taskId``) or update (``taskId="tsk_..."``) a scheduled task. + Keyword arguments become the JSON body: ``initiatorUserId`` (the caller or a + project service user, ``usr_...``), ``name``, ``cron`` (5 fields, UTC), + ``isEnabled``, ``stopOnError``, optional ``description`` and ``task``. + ``task`` is ``{"type": "EmailCampaign", "campaign": {...}, "databaseIntegrationId": ...}`` + — ``EmailCampaign`` is the only task type the gateway runs today; ``campaign`` + is the email campaign body (``source``, ``templateId`` and that source's + audience fields). See docs/hub/scheduler.md. + """ return self._transport.send( target="hub", path="/{version}/scheduler/tasks", @@ -119,11 +129,11 @@ def __init__(self, transport: AsyncTransport) -> None: self._transport = transport async def disable_scheduler(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """GET /{version}/scheduler/disable""" + """PUT /{version}/scheduler/disable""" return await self._transport.send( target="hub", path="/{version}/scheduler/disable", - method="GET", + method="PUT", path_params={}, request=request, scope="project", @@ -132,11 +142,11 @@ async def disable_scheduler(self, *, timeout: float | None = None, bearer_token: ) async def enable_scheduler(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """GET /{version}/scheduler/enable""" + """PUT /{version}/scheduler/enable""" return await self._transport.send( target="hub", path="/{version}/scheduler/enable", - method="GET", + method="PUT", path_params={}, request=request, scope="project", @@ -210,7 +220,17 @@ async def get_scheduler_tasks(self, *, timeout: float | None = None, bearer_toke ) async def save_scheduler_task(self, *, timeout: float | None = None, bearer_token: str | None = None, **request: Any) -> Any: - """POST /{version}/scheduler/tasks""" + """POST /{version}/scheduler/tasks + + Create (no ``taskId``) or update (``taskId="tsk_..."``) a scheduled task. + Keyword arguments become the JSON body: ``initiatorUserId`` (the caller or a + project service user, ``usr_...``), ``name``, ``cron`` (5 fields, UTC), + ``isEnabled``, ``stopOnError``, optional ``description`` and ``task``. + ``task`` is ``{"type": "EmailCampaign", "campaign": {...}, "databaseIntegrationId": ...}`` + — ``EmailCampaign`` is the only task type the gateway runs today; ``campaign`` + is the email campaign body (``source``, ``templateId`` and that source's + audience fields). See docs/hub/scheduler.md. + """ return await self._transport.send( target="hub", path="/{version}/scheduler/tasks", diff --git a/tests/hub/test_scheduler.py b/tests/hub/test_scheduler.py index deed314..859adaf 100644 --- a/tests/hub/test_scheduler.py +++ b/tests/hub/test_scheduler.py @@ -1,74 +1,206 @@ from __future__ import annotations -from norbix_python import Norbix, NorbixError +import asyncio +import json +from collections.abc import Callable +from typing import Any +from urllib.parse import parse_qs, urlparse + +import httpx +import pytest + +from norbix_python import AsyncNorbix + from ..helpers import make_client +"""Scheduler coverage. -def test_hub_scheduler_module_surface() -> None: - client, _ = make_client() - module = client.hub.scheduler - assert callable(module.disable_scheduler) - assert callable(module.enable_scheduler) - assert callable(module.delete_scheduler_task) - assert callable(module.disable_scheduler_task) - assert callable(module.enable_scheduler_task) - assert callable(module.get_scheduler_task) - assert callable(module.get_scheduler_tasks) - assert callable(module.save_scheduler_task) +Every scheduler method, sync and async: the verb, the fully-resolved path, and +where each parameter goes — the route token in the path, list filters in the +query string, the save request in the JSON body. Source of truth: gateway +Hub.Scheduler/*.cs [Route]. Every call goes to a capture transport, so nothing +leaves the process. +""" +TASK_ID = "tsk_1" +INITIATOR_USER_ID = "usr_1" +EMAIL_TEMPLATE_ID = "etpl_1" -def test_hub_scheduler_disable_scheduler_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.disable_scheduler() - assert transport.last_request['method'] == 'GET' - assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') +BASE = "/v2/scheduler" -def test_hub_scheduler_enable_scheduler_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.enable_scheduler() - assert transport.last_request['method'] == 'GET' - assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') +# A complete save request: the initiator, the schedule, and an email campaign +# task sent to all users. EmailCampaign is the only task type the gateway runs. +SAVE_REQUEST: dict[str, Any] = { + "initiatorUserId": INITIATOR_USER_ID, + "name": "Weekly digest", + "cron": "0 9 * * 1", + "isEnabled": True, + "stopOnError": False, + "task": { + "type": "EmailCampaign", + "campaign": {"source": "AllUsers", "templateId": EMAIL_TEMPLATE_ID}, + }, +} -def test_hub_scheduler_delete_scheduler_task_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.delete_scheduler_task(id="stub-Id") - assert transport.last_request['method'] == 'DELETE' - assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') +# name, verb, expected path, expected query (None: none), expected body (None: none), call +SchedulerCase = tuple[str, str, str, dict[str, list[str]] | None, dict[str, Any] | None, Callable[[Any], Any]] -def test_hub_scheduler_disable_scheduler_task_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.disable_scheduler_task(id="stub-Id") - assert transport.last_request['method'] == 'PUT' - assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') +SCHEDULER_CASES: list[SchedulerCase] = [ + # module — PUT since the gateway moved them off GET + ("enable_scheduler", "PUT", f"{BASE}/enable", None, None, lambda m: m.enable_scheduler()), + ("disable_scheduler", "PUT", f"{BASE}/disable", None, None, lambda m: m.disable_scheduler()), + # tasks + ( + "get_scheduler_tasks", + "GET", + f"{BASE}/tasks", + {"pageSize": ["10"], "type": ["EmailCampaign"]}, + None, + lambda m: m.get_scheduler_tasks(pageSize=10, type="EmailCampaign"), + ), + ( + "get_scheduler_task", + "GET", + f"{BASE}/tasks/{TASK_ID}", + None, + None, + lambda m: m.get_scheduler_task(id=TASK_ID), + ), + ( + "save_scheduler_task", + "POST", + f"{BASE}/tasks", + None, + SAVE_REQUEST, + lambda m: m.save_scheduler_task(**SAVE_REQUEST), + ), + ( + "enable_scheduler_task", + "PUT", + f"{BASE}/tasks/{TASK_ID}/enable", + None, + None, + lambda m: m.enable_scheduler_task(id=TASK_ID), + ), + ( + "disable_scheduler_task", + "PUT", + f"{BASE}/tasks/{TASK_ID}/disable", + None, + None, + lambda m: m.disable_scheduler_task(id=TASK_ID), + ), + ( + "delete_scheduler_task", + "DELETE", + f"{BASE}/tasks/{TASK_ID}", + None, + None, + lambda m: m.delete_scheduler_task(id=TASK_ID), + ), +] -def test_hub_scheduler_enable_scheduler_task_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.enable_scheduler_task(id="stub-Id") - assert transport.last_request['method'] == 'PUT' - assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') -def test_hub_scheduler_get_scheduler_task_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.get_scheduler_task(id="stub-id") - assert transport.last_request['method'] == 'GET' +def _assert_request( + name: str, + method: str, + url: str, + body: str, + verb: str, + path: str, + query: dict[str, list[str]] | None, + expected_body: dict[str, Any] | None, +) -> None: + parsed = urlparse(url) + assert method == verb, f"{name}: verb {method}, expected {verb}" + assert parsed.path == path, f"{name}: got {url}, expected the path {path}" + assert parse_qs(parsed.query) == (query or {}), f"{name}: query {parsed.query!r}" + if expected_body is None: + assert body == "", f"{name}: expected no body, got {body!r}" + else: + assert json.loads(body) == expected_body, f"{name}: body {body}" + + +@pytest.mark.parametrize( + ("name", "verb", "path", "query", "body", "call"), + SCHEDULER_CASES, + ids=[case[0] for case in SCHEDULER_CASES], +) +def test_scheduler_endpoint_hits_the_expected_route( + name: str, + verb: str, + path: str, + query: dict[str, list[str]] | None, + body: dict[str, Any] | None, + call: Callable[[Any], Any], +) -> None: + client, transport = make_client() + call(client.hub.scheduler) + assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') + _assert_request( + name, + transport.last_request["method"], + transport.last_request["url"], + transport.last_request["body"], + verb, + path, + query, + body, + ) + + +def test_async_scheduler_endpoints_hit_the_same_routes() -> None: + seen: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={}) + + async def run() -> None: + client = AsyncNorbix( + project_id="test-project", + bearer_token="test-token", + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)), + ) + try: + for _, _, _, _, _, call in SCHEDULER_CASES: + await call(client.hub.scheduler) + finally: + await client.aclose() + + asyncio.run(run()) + + assert len(seen) == len(SCHEDULER_CASES) + for request, (name, verb, path, query, body, _) in zip(seen, SCHEDULER_CASES, strict=True): + _assert_request(name, request.method, str(request.url), request.content.decode("utf-8"), verb, path, query, body) + + +def test_scheduler_surface_size() -> None: + """8 scheduler routes in the gateway. A new one changes this count, so it cannot arrive untested.""" + assert len(SCHEDULER_CASES) == 8 + assert len({case[0] for case in SCHEDULER_CASES}) == 8 + + +def test_save_scheduler_task_sends_the_email_campaign_task() -> None: + client, transport = make_client() + client.hub.scheduler.save_scheduler_task(**SAVE_REQUEST) -def test_hub_scheduler_get_scheduler_tasks_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.get_scheduler_tasks() - assert transport.last_request['method'] == 'GET' assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') + sent = json.loads(transport.last_request["body"]) + # The discriminator reaches the wire as the name; the campaign is nested under task. + assert sent["task"] == { + "type": "EmailCampaign", + "campaign": {"source": "AllUsers", "templateId": EMAIL_TEMPLATE_ID}, + } + assert sent["initiatorUserId"] == INITIATOR_USER_ID + assert sent["stopOnError"] is False + + +def test_save_scheduler_task_sends_the_task_id_in_the_body_on_update() -> None: + client, transport = make_client() + client.hub.scheduler.save_scheduler_task(taskId=TASK_ID, **SAVE_REQUEST) -def test_hub_scheduler_save_scheduler_task_request_shape() -> None: - client, transport = make_client(account_id=None) - client.hub.scheduler.save_scheduler_task() - assert transport.last_request['method'] == 'POST' assert transport.last_request is not None - assert transport.last_request['url'].startswith('https://') + assert urlparse(transport.last_request["url"]).path == f"{BASE}/tasks" + assert json.loads(transport.last_request["body"]) == {"taskId": TASK_ID, **SAVE_REQUEST}