def hourly_schedule(pipeline_name, start_date, name=None, execution_time=datetime.time(0, 0), tags_fn_for_date=None, solid_selection=None, mode='default', should_execute=None, environment_vars=None, end_date=None, execution_timezone=None):
    "Create a schedule that runs hourly.\n\n    The decorated function will be called as the ``run_config_fn`` of the underlying\n    :py:class:`~dagster.ScheduleDefinition` and should take a\n    :py:class:`~dagster.ScheduleExecutionContext` as its only argument, returning the environment\n    dict for the scheduled execution.\n\n    Args:\n        pipeline_name (str): The name of the pipeline to execute when the schedule runs.\n        start_date (datetime.datetime): The date from which to run the schedule.\n        name (Optional[str]): The name of the schedule to create. By default, this will be the name\n            of the decorated function.\n        execution_time (datetime.time): The time at which to execute the schedule. Only the minutes\n            component will be respected -- the hour should be 0, and will be ignored if it is not 0.\n        tags_fn_for_date (Optional[Callable[[datetime.datetime], Optional[Dict[str, str]]]]): A\n            function that generates tags to attach to the schedules runs. Takes the date of the\n            schedule run and returns a dictionary of tags (string key-value pairs).\n        solid_selection (Optional[List[str]]): A list of solid subselection (including single\n            solid names) to execute when the schedule runs. e.g. ``['*some_solid+', 'other_solid']``\n        mode (Optional[str]): The pipeline mode in which to execute this schedule.\n            (Default: 'default')\n        should_execute (Optional[Callable[ScheduleExecutionContext, bool]]): A function that runs at\n            schedule execution tie to determine whether a schedule should execute or skip. Takes a\n            :py:class:`~dagster.ScheduleExecutionContext` and returns a boolean (``True`` if the\n            schedule should execute). Defaults to a function that always returns ``True``.\n        environment_vars (Optional[Dict[str, str]]): Any environment variables to set when executing\n            the schedule.\n        end_date (Optional[datetime.datetime]): The last time to run the schedule to, defaults to\n            current time.\n        execution_timezone (Optional[str]): Timezone in which the schedule should run. Only works\n            with DagsterDaemonScheduler, and must be set when using that scheduler.\n    "
    check.opt_str_param(name, 'name')
    check.inst_param(start_date, 'start_date', datetime.datetime)
    check.opt_inst_param(end_date, 'end_date', datetime.datetime)
    check.opt_callable_param(tags_fn_for_date, 'tags_fn_for_date')
    check.opt_nullable_list_param(solid_selection, 'solid_selection', of_type=str)
    mode = check.opt_str_param(mode, 'mode', DEFAULT_MODE_NAME)
    check.opt_callable_param(should_execute, 'should_execute')
    check.opt_dict_param(environment_vars, 'environment_vars', key_type=str, value_type=str)
    check.str_param(pipeline_name, 'pipeline_name')
    check.inst_param(execution_time, 'execution_time', datetime.time)
    check.opt_str_param(execution_timezone, 'execution_timezone')
    if ((start_date.minute != 0) or (start_date.second != 0)):
        warnings.warn('`start_date` must be at the beginning of the hour for an hourly schedule. Use `execution_time` to execute the schedule at a specific time within the hour. For example, to run the schedule each hour at 15 minutes past the hour starting at 3AM on 10/20/2020, your schedule definition would look like:\n@hourly_schedule(\n    start_date=datetime.datetime(2020, 10, 20, 3),\n    execution_time=datetime.time(0, 15)\n):\ndef my_schedule_definition(_):\n    ...\n')
    if (execution_time.hour != 0):
        warnings.warn('Hourly schedule {schedule_name} created with:\n\tschedule_time=datetime.time(hour={hour}, minute={minute}, ...).Since this is an hourly schedule, the hour parameter will be ignored and the schedule will run on the {minute} mark for the previous hour interval. Replace datetime.time(hour={hour}, minute={minute}, ...) with datetime.time(minute={minute}, ...) to fix this warning.')
    cron_schedule = '{minute} * * * *'.format(minute=execution_time.minute)
    fmt = (DEFAULT_HOURLY_FORMAT_WITH_TIMEZONE if execution_timezone else DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE)
    execution_time_to_partition_fn = (lambda d: pendulum.instance(d).subtract(hours=1, minutes=((execution_time.minute - start_date.minute) % 60)))
    partition_fn = schedule_partition_range(start_date, end=end_date, cron_schedule=cron_schedule, fmt=fmt, timezone=execution_timezone, execution_time_to_partition_fn=execution_time_to_partition_fn)

    def inner(fn):
        check.callable_param(fn, 'fn')
        schedule_name = (name or fn.__name__)
        tags_fn_for_partition_value = (lambda partition: {})
        if tags_fn_for_date:
            tags_fn_for_partition_value = (lambda partition: tags_fn_for_date(partition.value))
        partition_set = PartitionSetDefinition(name='{}_partitions'.format(schedule_name), pipeline_name=pipeline_name, partition_fn=partition_fn, run_config_fn_for_partition=(lambda partition: fn(partition.value)), solid_selection=solid_selection, tags_fn_for_partition=tags_fn_for_partition_value, mode=mode)
        return partition_set.create_schedule_definition(schedule_name, cron_schedule, should_execute=should_execute, environment_vars=environment_vars, partition_selector=create_offset_partition_selector(execution_time_to_partition_fn=execution_time_to_partition_fn), execution_timezone=execution_timezone)
    return inner
Create a schedule that runs hourly.

The decorated function will be called as the ``run_config_fn`` of the underlying
:py:class:`~dagster.ScheduleDefinition` and should take a
:py:class:`~dagster.ScheduleExecutionContext` as its only argument, returning the environment
dict for the scheduled execution.

Args:
    pipeline_name (str): The name of the pipeline to execute when the schedule runs.
    start_date (datetime.datetime): The date from which to run the schedule.
    name (Optional[str]): The name of the schedule to create. By default, this will be the name
        of the decorated function.
    execution_time (datetime.time): The time at which to execute the schedule. Only the minutes
        component will be respected -- the hour should be 0, and will be ignored if it is not 0.
    tags_fn_for_date (Optional[Callable[[datetime.datetime], Optional[Dict[str, str]]]]): A
        function that generates tags to attach to the schedules runs. Takes the date of the
        schedule run and returns a dictionary of tags (string key-value pairs).
    solid_selection (Optional[List[str]]): A list of solid subselection (including single
        solid names) to execute when the schedule runs. e.g. ``['*some_solid+', 'other_solid']``
    mode (Optional[str]): The pipeline mode in which to execute this schedule.
        (Default: 'default')
    should_execute (Optional[Callable[ScheduleExecutionContext, bool]]): A function that runs at
        schedule execution tie to determine whether a schedule should execute or skip. Takes a
        :py:class:`~dagster.ScheduleExecutionContext` and returns a boolean (``True`` if the
        schedule should execute). Defaults to a function that always returns ``True``.
    environment_vars (Optional[Dict[str, str]]): Any environment variables to set when executing
        the schedule.
    end_date (Optional[datetime.datetime]): The last time to run the schedule to, defaults to
        current time.
    execution_timezone (Optional[str]): Timezone in which the schedule should run. Only works
        with DagsterDaemonScheduler, and must be set when using that scheduler.
def hourly_schedule(pipeline_name, start_date, name=None, execution_time=datetime.time(0, 0), tags_fn_for_date=None, solid_selection=None, mode='default', should_execute=None, environment_vars=None, end_date=None, execution_timezone=None):
    "Create a schedule that runs hourly.\n\n    The decorated function will be called as the ``run_config_fn`` of the underlying\n    :py:class:`~dagster.ScheduleDefinition` and should take a\n    :py:class:`~dagster.ScheduleExecutionContext` as its only argument, returning the environment\n    dict for the scheduled execution.\n\n    Args:\n        pipeline_name (str): The name of the pipeline to execute when the schedule runs.\n        start_date (datetime.datetime): The date from which to run the schedule.\n        name (Optional[str]): The name of the schedule to create. By default, this will be the name\n            of the decorated function.\n        execution_time (datetime.time): The time at which to execute the schedule. Only the minutes\n            component will be respected -- the hour should be 0, and will be ignored if it is not 0.\n        tags_fn_for_date (Optional[Callable[[datetime.datetime], Optional[Dict[str, str]]]]): A\n            function that generates tags to attach to the schedules runs. Takes the date of the\n            schedule run and returns a dictionary of tags (string key-value pairs).\n        solid_selection (Optional[List[str]]): A list of solid subselection (including single\n            solid names) to execute when the schedule runs. e.g. ``['*some_solid+', 'other_solid']``\n        mode (Optional[str]): The pipeline mode in which to execute this schedule.\n            (Default: 'default')\n        should_execute (Optional[Callable[ScheduleExecutionContext, bool]]): A function that runs at\n            schedule execution tie to determine whether a schedule should execute or skip. Takes a\n            :py:class:`~dagster.ScheduleExecutionContext` and returns a boolean (``True`` if the\n            schedule should execute). Defaults to a function that always returns ``True``.\n        environment_vars (Optional[Dict[str, str]]): Any environment variables to set when executing\n            the schedule.\n        end_date (Optional[datetime.datetime]): The last time to run the schedule to, defaults to\n            current time.\n        execution_timezone (Optional[str]): Timezone in which the schedule should run. Only works\n            with DagsterDaemonScheduler, and must be set when using that scheduler.\n    "
    check.opt_str_param(name, 'name')
    check.inst_param(start_date, 'start_date', datetime.datetime)
    check.opt_inst_param(end_date, 'end_date', datetime.datetime)
    check.opt_callable_param(tags_fn_for_date, 'tags_fn_for_date')
    check.opt_nullable_list_param(solid_selection, 'solid_selection', of_type=str)
    mode = check.opt_str_param(mode, 'mode', DEFAULT_MODE_NAME)
    check.opt_callable_param(should_execute, 'should_execute')
    check.opt_dict_param(environment_vars, 'environment_vars', key_type=str, value_type=str)
    check.str_param(pipeline_name, 'pipeline_name')
    check.inst_param(execution_time, 'execution_time', datetime.time)
    check.opt_str_param(execution_timezone, 'execution_timezone')
    if ((start_date.minute != 0) or (start_date.second != 0)):
        warnings.warn('`start_date` must be at the beginning of the hour for an hourly schedule. Use `execution_time` to execute the schedule at a specific time within the hour. For example, to run the schedule each hour at 15 minutes past the hour starting at 3AM on 10/20/2020, your schedule definition would look like:\n@hourly_schedule(\n    start_date=datetime.datetime(2020, 10, 20, 3),\n    execution_time=datetime.time(0, 15)\n):\ndef my_schedule_definition(_):\n    ...\n')
    if (execution_time.hour != 0):
        warnings.warn('Hourly schedule {schedule_name} created with:\n\tschedule_time=datetime.time(hour={hour}, minute={minute}, ...).Since this is an hourly schedule, the hour parameter will be ignored and the schedule will run on the {minute} mark for the previous hour interval. Replace datetime.time(hour={hour}, minute={minute}, ...) with datetime.time(minute={minute}, ...) to fix this warning.')
    cron_schedule = '{minute} * * * *'.format(minute=execution_time.minute)
    fmt = (DEFAULT_HOURLY_FORMAT_WITH_TIMEZONE if execution_timezone else DEFAULT_HOURLY_FORMAT_WITHOUT_TIMEZONE)
    execution_time_to_partition_fn = (lambda d: pendulum.instance(d).subtract(hours=1, minutes=((execution_time.minute - start_date.minute) % 60)))
    partition_fn = schedule_partition_range(start_date, end=end_date, cron_schedule=cron_schedule, fmt=fmt, timezone=execution_timezone, execution_time_to_partition_fn=execution_time_to_partition_fn)

    def inner(fn):
        check.callable_param(fn, 'fn')
        schedule_name = (name or fn.__name__)
        tags_fn_for_partition_value = (lambda partition: {})
        if tags_fn_for_date:
            tags_fn_for_partition_value = (lambda partition: tags_fn_for_date(partition.value))
        partition_set = PartitionSetDefinition(name='{}_partitions'.format(schedule_name), pipeline_name=pipeline_name, partition_fn=partition_fn, run_config_fn_for_partition=(lambda partition: fn(partition.value)), solid_selection=solid_selection, tags_fn_for_partition=tags_fn_for_partition_value, mode=mode)
        return partition_set.create_schedule_definition(schedule_name, cron_schedule, should_execute=should_execute, environment_vars=environment_vars, partition_selector=create_offset_partition_selector(execution_time_to_partition_fn=execution_time_to_partition_fn), execution_timezone=execution_timezone)
    return inner