Schedules are not created dynamically ( from seonsor ? )

[moxgeek]

im trying to create a schedule dynamically based on a trigger ( which is a sensor event ).
the code shows that everything is working fine, no exception, even all logs are going well.
except that no schedules has been created.

@sensor(name="redis_pipeline_sensor", jobs=load_pipelines())
def redis_pipeline_sensor(context):
    """
    Sensor that polls Redis for pipeline trigger messages and dynamically triggers jobs.
    """
    # Check for a message in the Redis queue
    message = redis_client.lpop("dagster_pipeline_queue")

    if message:
        context.log.info("✅ Message received from Redis")
        message_data = json.loads(message)

        # Extract config and derive job_name
        run_config = message_data.get("config", {})
        ops = run_config.get("ops", {})
        if len(ops) != 1:
            context.log.error("No valid ops found in config to derive job_name. Skipping.")
            return

        op_name = next(iter(ops))  # Get the first (and assumed only) key in `ops`
        if not op_name.startswith("fetch_"):
            context.log.error("Unable to derive job_name from op name. Skipping.")
            return

        job_name = op_name.replace("fetch_", "") + "_pipeline"
        context.log.info(f"Derived job_name '{job_name}' from op '{op_name}'.")

        # Dynamically find the job by name
        pipelines = {job.name: job for job in load_pipelines()}

        if job_name not in pipelines:
            context.log.error(f"Job '{job_name}' not found. Skipping.")
            return

        # Handle scheduling logic
        schedule_cron = message_data.get("schedule_cron")  # New: Cron schedule for the job
        run_key = message_data.get("run_key", None)

        if schedule_cron:
            context.log.info(f"Creating a schedule for job '{job_name}' with cron '{schedule_cron}'.")

            # Create a schedule definition and add it to the global registry
            def schedule_fn(_context):
                return RunRequest(run_key=run_key, run_config=run_config)

            dynamic_schedule = ScheduleDefinition(
                name=f"{job_name}_schedule",
                cron_schedule=schedule_cron,
                job=pipelines[job_name],
                run_config_fn=schedule_fn,
            )
            dynamic_schedules.append(dynamic_schedule)

            context.log.info(f"Schedule created for job '{job_name}' with cron '{schedule_cron}'.")
        else:
            # Trigger the job immediately
            context.log.info(f"Triggering job '{job_name}' immediately.")
            yield RunRequest(run_key=run_key, run_config=run_config, job_name=job_name)

To explain my requirement, here’s the context:
I already have several jobs defined, and I can trigger them using sensors. However, I need the ability to schedule a job dynamically with a specific configuration.

Example:

I have a job called fetch_news, which reads news from an RSS source.

One source updates once a day, so the job should run daily.

Another source updates three times a day, so the same job should run with a different schedule.

My sources can scale over time, meaning new sources and their corresponding schedules can be added dynamically.

Currently, creating a separate schedule for each job/configuration manually (or hardcoding them) doesn’t feel scalable or efficient.

What I’d like to achieve is:

Be able to dynamically send new configurations, including the schedule (e.g., a cron expression) from an external system via sensors such as a publisher in Redis as my example.

Dagster would then automatically create or adjust the corresponding schedule for the job without requiring manual intervention or predefined schedules.

if it's possible and with the vision design of dagster need to be shown on the list of schedules and the UI.
an error if it's not possible ( as the graphQL ... ).

dynamic_schedules is a global variable that returned by the repository.

dagster, version 1.9.3

[garethbrickman]

I've never seen a sensor dynamically create a schedule definition. I think schedule definitions must be present in the user code before Dagster is deployed.

[moxgeek]

To explain my requirement, here’s the context:
I already have several jobs defined, and I can trigger them using sensors. However, I need the ability to schedule a job dynamically with a specific configuration.

Example:

I have a job called fetch_news, which reads news from an RSS source.

One source updates once a day, so the job should run daily.

Another source updates three times a day, so the same job should run with a different schedule.

My sources can scale over time, meaning new sources and their corresponding schedules can be added dynamically.

Currently, creating a separate schedule for each job/configuration manually (or hardcoding them) doesn’t feel scalable or efficient.

What I’d like to achieve is:

Be able to dynamically send new configurations, including the schedule (e.g., a cron expression) from an external system via sensors such as a publisher in Redis as my example.

Dagster would then automatically create or adjust the corresponding schedule for the job without requiring manual intervention or predefined schedules.