One-time scheduling

Benefits

• Precise timing: Unlike a delayed message, a one-time task is not constrained by fixed delay periods. You can schedule it for a specific future time, simplifying the configuration.

• Versatile task support: one-time scheduling works with all task types, including Java, HTTP, and Shell. It also supports all distributed models, such as standalone, broadcast, sharding, and MapReduce.

• Visual management: Like other tasks in SchedulerX, one-time tasks can be monitored and queried via a visual interface. You can modify task parameters before the schedule time and enable features like automatic retries on failure.

Configure one-time scheduling

Select one-time scheduling when you create a task.

1. Navigate to the Timing configuration panel. For more information, see Create a task.

2. In the Add Task wizard, complete the basic settings and click Next to open the Timing configuration panel.

3. For Time type, select one_time from the list, and then specify the Scheduling time.

   Optional: In the Advanced Settings section, you can configure the Time Offset, Time zone, and Effective Time.

   Parameter	Description	Default
   Data Timestamp Offset	The offset for the data timestamp, in seconds.	None
   Time zone	The time zone for scheduling the task.	None
   Effective Time	The time when the task becomes active.	Immediately

View tasks

1. Log on to the MSE SchedulerX console, and select a region in the top navigation bar.

2. In the navigation pane on the left, click Task Management to view your scheduled tasks.

Create a one-time task using the API

The following example shows how to use the SDK to create a one-time task. To enable one-time scheduling, set the timeType parameter to 5. For more information, see the OpenAPI portal.

import os
import sys

from typing import List

from alibabacloud_schedulerx220190430.client import Client as schedulerx220190430Client
from alibabacloud_credentials.client import Client as CredentialClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_schedulerx220190430 import models as schedulerx_220190430_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient


class Sample:
    def __init__(self):
        pass

    @staticmethod
    def create_client() -> schedulerx220190430Client:
        """
        Initializes a client with credentials.
        @return: Client
        @throws Exception
        """
        # For production environments, we recommend using a more secure authentication method, such as RAM roles. For more information about credential configuration, see https://www.alibabacloud.com/help/document_detail/378659.html.                       
        config = open_api_models.Config(
            access_key_id=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID'),
            access_key_secret=os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET')
        )
        # For a list of available endpoints, see https://api.aliyun.com/product/schedulerx2.
        config.endpoint = f'schedulerx.aliyuncs.com'
        return schedulerx220190430Client(config)

    @staticmethod
    def main(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_job_request = schedulerx_220190430_models.CreateJobRequest(
            region_id='cn-hangzhou',
            time_type=5, # Enable one-time scheduling.
            namespace='YOUR_NAMESPACE',
            group_id='YOUR_GROUP_ID',
            job_type='python',
            name='testjob',
            execute_mode='standalone',
            time_expression='2025-10-10 10:10:00', # The time when the task is scheduled to run.
            class_name='com.alibaba.schedulerx.test.helloworld'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # When you run the code, print the API response to view the result.
            client.create_job_with_options(create_job_request, runtime)
        except Exception as error:
            # This is for demonstration only. In a production environment, handle exceptions properly and do not ignore them.
            # Error message
            print(error.message)
            # Diagnostic address
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)

    @staticmethod
    async def main_async(
        args: List[str],
    ) -> None:
        client = Sample.create_client()
        create_job_request = schedulerx_220190430_models.CreateJobRequest(
            region_id='cn-hangzhou',
            time_type=5, # Enable one-time scheduling.
            namespace='YOUR_NAMESPACE',
            group_id='YOUR_GROUP_ID',
            job_type='python',
            name='testjob',
            execute_mode='standalone',
            time_expression='2025-10-10 10:10:00', # The time when the task is scheduled to run.
            class_name='com.alibaba.schedulerx.test.helloworld'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # When you run the code, print the API response to view the result.
            await client.create_job_with_options_async(create_job_request, runtime)
        except Exception as error:
            # This is for demonstration only. In a production environment, handle exceptions properly and do not ignore them.
            # Error message
            print(error.message)
            # Diagnostic address
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])