navidrome/plugins/pdk/python/host/nd_host_scheduler.py

# Code generated by ndpgen. DO NOT EDIT.
#
# This file contains client wrappers for the Scheduler host service.
# It is intended for use in Navidrome plugins built with extism-py.
#
# IMPORTANT: Due to a limitation in extism-py, you cannot import this file directly.
# The @extism.import_fn decorators are only detected when defined in the plugin's
# main __init__.py file. Copy the needed functions from this file into your plugin.

from dataclasses import dataclass
from typing import Any

import extism
import json


class HostFunctionError(Exception):
    """Raised when a host function returns an error."""
    pass


@extism.import_fn("extism:host/user", "scheduler_scheduleonetime")
def _scheduler_scheduleonetime(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "scheduler_schedulerecurring")
def _scheduler_schedulerecurring(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "scheduler_cancelschedule")
def _scheduler_cancelschedule(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


def scheduler_schedule_one_time(delay_seconds: int, payload: str, schedule_id: str) -> str:
    """ScheduleOneTime schedules a one-time event to be triggered after the specified delay.
Plugins that use this function must also implement the SchedulerCallback capability

Parameters:
  - delaySeconds: Number of seconds to wait before triggering the event
  - payload: Data to be passed to the scheduled event handler
  - scheduleID: Optional unique identifier for the scheduled job. If empty, one will be generated

Returns the schedule ID that can be used to cancel the job, or an error if scheduling fails.

    Args:
        delay_seconds: int parameter.
        payload: str parameter.
        schedule_id: str parameter.

    Returns:
        str: The result value.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "delaySeconds": delay_seconds,
        "payload": payload,
        "scheduleId": schedule_id,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _scheduler_scheduleonetime(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])

    return response.get("newScheduleId", "")


def scheduler_schedule_recurring(cron_expression: str, payload: str, schedule_id: str) -> str:
    """ScheduleRecurring schedules a recurring event using a cron expression.
Plugins that use this function must also implement the SchedulerCallback capability

Parameters:
  - cronExpression: Standard cron format expression (e.g., "0 0 * * *" for daily at midnight)
  - payload: Data to be passed to each scheduled event handler invocation
  - scheduleID: Optional unique identifier for the scheduled job. If empty, one will be generated

Returns the schedule ID that can be used to cancel the job, or an error if scheduling fails.

    Args:
        cron_expression: str parameter.
        payload: str parameter.
        schedule_id: str parameter.

    Returns:
        str: The result value.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "cronExpression": cron_expression,
        "payload": payload,
        "scheduleId": schedule_id,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _scheduler_schedulerecurring(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])

    return response.get("newScheduleId", "")


def scheduler_cancel_schedule(schedule_id: str) -> None:
    """CancelSchedule cancels a scheduled job identified by its schedule ID.

This works for both one-time and recurring schedules. Once cancelled, the job will not trigger
any future events.

Returns an error if the schedule ID is not found or if cancellation fails.

    Args:
        schedule_id: str parameter.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "scheduleId": schedule_id,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _scheduler_cancelschedule(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])