CRUD API calls for scheduler (#1214)

* add linker table for workflow apscheduler

* refactor for apschedule

* Add crud API for creating schedule jobs

* Update after review

* remove export

* undo psql

* fix linting

* remove unused

* working example

* Refactor and update Graphql

* update settings

* testing added

* Added migrations for existing schedules

* fixing unit tests

* update get

* update settings

* update docs

* Fix linting and test

* fix unit tests and mypy

* fix pre-commit

* add more tests

* update after review

* refactor after review

* restore validate_products file

* fix minor issues

* add upgrading guide for schedules

* fix linting'

Author: Igoranze

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 4.6.5
+current_version = 4.7.0rc1
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(rc(?P<build>\d+))?

docs/guides/tasks.md

@@ -112,7 +112,9 @@ Even if the task does not have any form input, an entry will still need to be ma
 }
 ```
 
-## The schedule file
+## The schedule file (DEPRECATED)
+> This section is deprecated and will be removed in version 5.0.0, please refer to the [new scheduling system](#the-schedule-api)
+> below.
 
 > from `4.3.0` we switched from [schedule] package to [apscheduler] to allow schedules to be stored in the DB and schedule tasks from the API.
 
@@ -152,11 +154,97 @@ To keep things organized and consistent (similar to how workflows are handled),
 > In previous versions, schedules needed to be explicitly listed in an ALL_SCHEDULERS variable.
 > This is no longer required, but ALL_SCHEDULERS is still supported for backwards compatibility.
 
+
+## The schedule API
+
+> from `4.3.0` we switched from [schedule] package to [apscheduler] to allow schedules to be stored in the DB and schedule tasks from the API.
+
+> from `4.7.0` we deprecated `@scheduler.scheduled_job()` provided by [apscheduler] in favor of a more dynamic API based system.
+> Although we do no longer support the `@scheduler.scheduled_job()` decorator, it is still available because it is part of [apscheduler].
+> Therefore, we do NOT recommend using it for new schedules. Because you will miss a Linker Table join between schedules and workflows/tasks.
+
+
+Schedules can be created, updated, and deleted via the REST API, and retrieved via the already existing GraphQL API. It
+will become possible to manage schedules through the
+UI ([development ticket](https://github.com/workfloworchestrator/orchestrator-ui-library/issues/2215)), but you may also
+use the API directly to automate configuration of your schedules.
+
+*Example POST*
+
+To create a schedule, you can now simply run a `POST` request to the `/api/schedules` endpoint with a JSON body containing the schedule details.
+An example body to create a nightly sync schedule would look like this:
+
+```json
+{
+  "name": "Nightly Product Validation",
+  "workflow_name": "validate_products",
+  "workflow_id": "e96cc6bb-9494-4ac1-a572-050988487ee1",
+  "trigger": "interval",
+  "trigger_kwargs": {
+    "hours": 12
+  }
+}
+```
+
+Respectively, you can update or delete schedules via `PUT` and `DELETE` requests to the same endpoint.
+
+*Example PUT*
+
+With `PUT` you can only update the `name`, `trigger`, and `trigger_kwargs` of an existing schedule.
+For example, to update the above schedule to run every 24 hours instead of every 12 hours
+```json
+{
+  "schedule_id": "c1b6e5e3-d9f0-48f2-bc65-3c9c33fcf561",
+  "name": "Updated Nightly Cleanup",
+  "trigger": "interval",
+  "trigger_kwargs": {
+    "hours": 24
+  }
+}
+```
+
+*Example DELETE*
+
+To delete a schedule, you only need to provide the `schedule_id` in the `DELETE` call
+```json
+{
+  "workflow_id": "b67d4ca7-19fb-4b83-a022-34c6322fb5f1",
+  "schedule_id": "1fe43a96-b0f4-4c89-b9b7-87db14bbd8d3"
+}
+```
+
+There are multiple triggers that can be used ([trigger docs]):
+
+- [IntervalTrigger]: use when you want to run the task at fixed intervals of time.
+- [CronTrigger]: use when you want to run the task periodically at certain time(s) of day.
+- [DateTrigger]: use when you want to run the task just once at a certain point of time.
+- [CalendarIntervalTrigger]: use when you want to run the task on calendar-based intervals, at a specific time of day.
+- [AndTrigger]: use when you want to combine multiple triggers so the task only runs when **all** of them would fire at the same time.
+- [OrTrigger]: use when you want to combine multiple triggers so the task runs when **any one** of them would fire.
+
+For detailed configuration options, see the [APScheduler scheduling docs].
+
+The scheduler automatically loads any schedules that are imported before the scheduler starts.
+
+> In previous versions, schedules needed to be explicitly listed in an ALL_SCHEDULERS variable.
+> This is no longer required; ALL_SCHEDULERS is deprecated as of orchestrator-core 4.7.0 and will be removed in 5.0.0.
+> Follow-up ticket to remove deprecated code: [#1276](https://github.com/workfloworchestrator/orchestrator-core/issues/1276)
+
 ## The scheduler
 
 The scheduler is invoked via `python main.py scheduler`.
 Try `--help` or review the [CLI docs][cli-docs] to learn more.
 
+### Initial schedules
+From version orchestrator-core >= `4.7.0`, the scheduler uses the database to store schedules instead of hard-coded schedule files.
+Previous versions (orchestrator-core < `4.7.0` had hard-coded schedules. These can be ported to the new system by creating them via the API or CLI.
+Run the following CLI command to import previously existing orchestrator-core schedules and change them if needed via the API.
+
+```shell
+python main.py scheduler load-initial-schedule
+```
+
+> Remember, that if you do not explicitly import these, they will not be available to the scheduler.
 ### Manually executing tasks
 
 When doing development, it is possible to manually make the scheduler run your task even if your Orchestrator instance is not in "scheduler mode."
@@ -165,11 +253,11 @@ Shell into your running instance and run the following:
 
 ```shell
 docker exec -it backend /bin/bash
-python main.py scheduler force run_nightly_sync
+python main.py scheduler force "c1b6e5e3-d9f0-48f2-bc65-3c9c33fcf561"
 ```
 
-...where `run_nightly_sync` is the job defined in the schedule file - not the name of the task.
-This doesn't depend on the UI being up, and you can get the logging output.
+...where `c1b6e5e3-d9f0-48f2-bc65-3c9c33fcf561` is the job id of the schedule you want to run.
+The job id can be found via the GraphQL API or directly in the database.
 
 ### Starting the scheduler
 
@@ -213,30 +301,6 @@ fi
 ```
 
 
-## Developer notes
-
-### Executing multiple tasks
-
-If one needs to execute multiple tasks in concert with each other, one can not call a task from another task. Which is to say, calling `start_process` is a "top level" call. Trying to call it inside an already invoked task does not work.
-
-But the schedule (ie: crontab) files are also code modules so one can achieve the same thing there:
-
-```python
-@scheduler(name="Nightly sync", time_unit="day", at="00:10")
-def run_nightly_sync() -> None:
-    subs = Subscription.query.filter(
-        Subscription.description.like("Node%Provisioned")
-    ).all()
-    logger.info("Node schedule subs", subs=subs)
-
-    for sub in subs:
-        sub_id = sub.subscription_id
-        logger.info("Validate node enrollment", sub_id=sub_id)
-        start_process("validate_node_enrollment", [{"subscription_id": sub_id}])
-
-    start_process("task_sync_from")
-```
-
 [schedule]: https://pypi.org/project/schedule/
 [apscheduler]: https://pypi.org/project/APScheduler/
 [IntervalTrigger]: https://apscheduler.readthedocs.io/en/master/api.html#apscheduler.triggers.interval.IntervalTrigger

docs/guides/upgrading/4.0.md

@@ -71,3 +71,26 @@ This will update the `target` for all workflows that are `SYSTEM` or `VALIDATE`
 
 This is a breaking change, so you will need to test your workflows after making this change to ensure that they are
 working as expected.
+
+## Scheduling flows
+In 4.4.0 we have introduced a new way to schedule workflows via the decorator [@scheduler.scheduled_job(...)](../tasks.md#the-schedule-file-deprecated).
+In 4.7.0 we introduce a new way to schedule workflows via the [Scheduler API](../tasks.md#the-schedule-api).
+
+The 4.7.0 migration will create the `apscheduler_jobs` table if missing, so upgrades from older versions still succeed, but the table will initially be empty.
+
+### Who needs to take action?
+
+**Users upgrading from 4.4–4.6:**
+ - Nothing special required; your scheduler data already exists.
+
+**Users upgrading from <4.4 who used the old decorator:**
+ - Your schedules will not automatically reappear. Restore them by running:
+ -
+   ```
+   python main.py scheduler load-initial-schedule
+   ```
+   (More information: [Scheduler API](../tasks.md#the-schedule-api).)
+   Make sure your schedule definitions are imported so the scheduler can discover them.
+
+**Users who never used the decorator:**
+- No action required; the migration creates the table and you can add schedules normally.

docs/reference-docs/websockets.md

@@ -18,7 +18,7 @@ The main component is the `WebSocketManager` (WSM) which has the following respo
 3. Provide an interface to pass messages from a backend process (workflow/task)
 
 In a setup with multiple isolated Orchestrator instances the WSM is initialized multiple times as well, therefore clients can be connected to any arbitrary WSM instance.
-Letting a backend process broadcast messages to all clients thus requires a message broker, for which we use [Redis Pub/Sub](https://redis.io/docs/manual/pubsub).
+Letting a backend process broadcast messages to all clients thus requires a message broker, for which we use [Redis Pub/Sub](https://redis.io/docs/latest/develop/pubsub/).
 
 There are 2 WSM implementations: a `MemoryWebsocketManager` for development/testing, and a `BroadcastWebsocketManager` that connects to Redis. We'll continue to discuss the latter.
 

orchestrator/__init__.py

@@ -13,7 +13,7 @@
 
 """This is the orchestrator workflow engine."""
 
-__version__ = "4.6.5"
+__version__ = "4.7.0rc1"
 
 
 from structlog import get_logger

orchestrator/api/api_v1/api.py

@@ -22,6 +22,7 @@
     product_blocks,
     products,
     resource_types,
+    schedules,
     settings,
     subscription_customer_descriptions,
     subscriptions,
@@ -88,6 +89,9 @@
 api_router.include_router(
     ws.router, prefix="/ws", tags=["Core", "Events"]
 )  # Auth on the websocket is handled in the Websocket Manager
+api_router.include_router(
+    schedules.router, prefix="/schedules", tags=["Core", "Schedules"], dependencies=[Depends(authorize)]
+)
 
 if llm_settings.SEARCH_ENABLED:
     from orchestrator.api.api_v1.endpoints import search

orchestrator/api/api_v1/endpoints/schedules.py

@@ -0,0 +1,44 @@
+# Copyright 2019-2025 SURF.
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+from http import HTTPStatus
+
+import structlog
+from fastapi.routing import APIRouter
+
+from orchestrator.schedules.service import add_scheduled_task_to_queue
+from orchestrator.schemas.schedules import APSchedulerJobCreate, APSchedulerJobDelete, APSchedulerJobUpdate
+
+logger = structlog.get_logger(__name__)
+
+router: APIRouter = APIRouter()
+
+
+@router.post("/", status_code=HTTPStatus.CREATED)
+def create_scheduled_task(payload: APSchedulerJobCreate) -> dict[str, str]:
+    """Create a scheduled task."""
+    add_scheduled_task_to_queue(payload)
+    return {"message": "Added to Create Queue", "status": "CREATED"}
+
+
+@router.put("/", status_code=HTTPStatus.OK)
+async def update_scheduled_task(payload: APSchedulerJobUpdate) -> dict[str, str]:
+    """Update a scheduled task."""
+    add_scheduled_task_to_queue(payload)
+    return {"message": "Added to Update Queue", "status": "UPDATED"}
+
+
+@router.delete("/", status_code=HTTPStatus.OK)
+async def delete_scheduled_task(payload: APSchedulerJobDelete) -> dict[str, str]:
+    """Delete a scheduled task."""
+    add_scheduled_task_to_queue(payload)
+    return {"message": "Added to Delete Queue", "status": "DELETED"}

orchestrator/cli/scheduler.py

@@ -10,28 +10,56 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
-
 import time
+from typing import cast
 
 import typer
+from redis import Redis
 
 from orchestrator.schedules.scheduler import (
     get_all_scheduler_tasks,
     get_scheduler,
     get_scheduler_task,
 )
+from orchestrator.schedules.service import (
+    SCHEDULER_QUEUE,
+    add_scheduled_task_to_queue,
+    workflow_scheduler_queue,
+)
+from orchestrator.schemas.schedules import APSchedulerJobCreate
+from orchestrator.services.workflows import get_workflow_by_name
+from orchestrator.settings import app_settings
+from orchestrator.utils.redis_client import create_redis_client
 
 app: typer.Typer = typer.Typer()
 
 
 @app.command()
 def run() -> None:
     """Start scheduler and loop eternally to keep thread alive."""
-    with get_scheduler():
-        while True:
+
+    def _get_scheduled_task_item_from_queue(redis_conn: Redis) -> tuple[str, bytes] | None:
+        """Get an item from the Redis Queue for scheduler tasks."""
+        try:
+            return redis_conn.brpop(SCHEDULER_QUEUE, timeout=1)
+        except ConnectionError as e:
+            typer.echo(f"There was a connection error with Redis. Retrying in 3 seconds... {e}")
+            time.sleep(3)
+        except Exception as e:
+            typer.echo(f"There was an unexpected error with Redis. Retrying in 1 second... {e}")
             time.sleep(1)
 
+        return None
+
+    with get_scheduler() as scheduler_connection:
+        redis_connection = create_redis_client(app_settings.CACHE_URI)
+        while True:
+            item = _get_scheduled_task_item_from_queue(redis_connection)
+            if not item:
+                continue
+
+            workflow_scheduler_queue(item, scheduler_connection)
+
 
 @app.command()
 def show_schedule() -> None:
@@ -59,3 +87,45 @@ def force(task_id: str) -> None:
     except Exception as e:
         typer.echo(f"Task execution failed: {e}")
         raise typer.Exit(code=1)
+
+
+@app.command()
+def load_initial_schedule() -> None:
+    """Load the initial schedule into the scheduler."""
+    initial_schedules = [
+        {
+            "name": "Task Resume Workflows",
+            "workflow_name": "task_resume_workflows",
+            "workflow_id": "",
+            "trigger": "interval",
+            "trigger_kwargs": {"hours": 1},
+        },
+        {
+            "name": "Task Clean Up Tasks",
+            "workflow_name": "task_clean_up_tasks",
+            "workflow_id": "",
+            "trigger": "interval",
+            "trigger_kwargs": {"hours": 6},
+        },
+        {
+            "name": "Task Validate Subscriptions",
+            "workflow_name": "task_validate_subscriptions",
+            "workflow_id": "",
+            "trigger": "cron",
+            "trigger_kwargs": {"hour": 0, "minute": 10},
+        },
+    ]
+
+    for schedule in initial_schedules:
+        # enrich with workflow id
+        workflow_name = cast(str, schedule.get("workflow_name"))
+        workflow = get_workflow_by_name(workflow_name)
+
+        if not workflow:
+            typer.echo(f"Workflow '{schedule['workflow_name']}' not found. Skipping schedule.")
+            continue
+
+        schedule["workflow_id"] = workflow.workflow_id
+
+        typer.echo(f"Initial Schedule: {schedule}")
+        add_scheduled_task_to_queue(APSchedulerJobCreate(**schedule))  # type: ignore