Scheduler (autotransform.util.scheduler)

Provides utility methods for interacting with the scheduling of AutoTransform.

class autotransform.util.scheduler.RepeatSetting(value)

Bases: str, Enum

Possible repitition schedules.

DAILY = 'daily'

WEEKLY = 'weekly'

class autotransform.util.scheduler.ScheduledSchema(*, schema_name: str, schedule: SchemaScheduleSettings, max_submissions: int | None = None)

Bases: ComponentModel

A Schema that is scheduled for automatic runs.

schema_name

The Schema that is being scheduled.

Type:: str

schedule

The settings used to determine when to run the Schema.

Type:: SchemaScheduleSettings

max_submissions

The max number of submissions from a given scheduled run. No limit if None. Defaults to None.

Type:: Optional[int], optional

static from_console() → ScheduledSchema

Gets a ScheduledSchema from console inputs.

Returns:: The input ScheduledSchema.
Return type:: ScheduledSchema

static from_data(data: Dict[str, Any]) → ScheduledSchema

Produces an instance of the ScheduledSchema from decoded data.

Parameters:: data (Mapping[str, Any]) – The JSON decoded data.
Returns:: An instance of the ScheduledSchema.
Return type:: ScheduledSchema

max_submissions: int | None

classmethod max_submissions_is_positive(v: int | None) → int | None

Validates that max submissions is positive.

Parameters:

cls (Type[ScheduledSchema]) – The ScheduledSchema class.
v (int) – The maximum number of submissions.

Raises:

ValueError – Raised if the maximum number of submissions is not positive.

Returns:

The unmodified maximum number of submissions.

Return type:

Optional[int]

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

schedule: SchemaScheduleSettings

schema_name: str

class autotransform.util.scheduler.Scheduler(*, base_time: int, excluded_days: List[int], schemas: List[ScheduledSchema])

Bases: ComponentModel

The information and functionality required to schedule Schemas.

base_time

The base time to use when determining hour_of_day, day_of_week, and valid shards. Considered day 0, hour 0.

Type:: int

excluded_days

A list of days of the week to skip running Schemas.

Type:: List[int]

schemas

A list of Schemas to schedule.

Type:: List[ScheduledSchema]

base_time: int

excluded_days: List[int]

static from_console(prev_scheduler: Scheduler | None = None) → Scheduler

Gets a Scheduler using console inputs.

Parameters:: prev_scheduler (Optional[Scheduler], optional) – A previously input Scheduler. Defaults to None.
Returns:: The input Scheduler.
Return type:: Scheduler

static from_data(data: Dict[str, Any]) → Scheduler

Produces an instance of the Scheduler from decoded data.

Parameters:: data (Mapping[str, Any]) – The JSON decoded data.
Returns:: An instance of the Scheduler.
Return type:: Scheduler

static from_json(scheduler_json: str) → Scheduler

Builds a Scheduler from JSON encoded values.

Parameters:

scheduler_json (str) – The JSON encoded Scheduler.
start_time (int) – The start time to use for setting up sharding/running.

Returns:

The Scheduler from the JSON.

Return type:

Scheduler

static init_from_console(use_sample_schema: bool = False, simple: bool = False) → Scheduler

Gets a Scheduler using console input.

Parameters:

use_sample_schema (bool, optional) – Whether to include the sample Schema. Defaults to False.
simple (bool, optional) – Whether to use simple settings. Defaults to False.

Returns:

The input Scheduler.

Return type:

Scheduler

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

static read(file_path: str) → Scheduler

Reads a Scheduler from a JSON encoded file.

Parameters:: file_path (str) – The path where the JSON for the Schedule is located.
Returns:: The Scheduler from the file.
Return type:: Scheduler

run(start_time: int, runner: Runner) → None

Runs the schedule.

Parameters:: start_time (int) – The time that the schedule is run on.

schemas: List[ScheduledSchema]

write(file_path: str) → None

Writes the Scheduler to a file as JSON.

Parameters:: file_path (str) – The file to write to.

class autotransform.util.scheduler.SchemaScheduleSettings(*, repeats: RepeatSetting, hour_of_day: int = 0, day_of_week: int | None = 0, shard_filter: ShardFilter | None = None)

Bases: ComponentModel

A schedule for a Schema.

repeats

How often the Schema will be run, daily or weekly.

Type:: RepeatSetting

hour_of_day

The hour of the day to run the Schema, 0-23. Defaults to 0.

Type:: int, optional

day_of_week

The day of the week to run a weekly Schema. Defaults to 0.

Type:: Optional[int], optional

shard_filter

A shard filter to use for sharding the inputs across multiple runs. Defaults to None.

Type:: Optional[ShardingSettings], optional

day_of_week: int | None

static from_console(simple: bool = False) → SchemaScheduleSettings

Gets a SchemaScheduleSettings from console inputs.

simple

Whether to use simple settings. Defaults to False.

Type:: bool, optional

Returns:: The input SchemaScheduleSettings.
Return type:: SchemaScheduleSettings

static from_data(data: Dict[str, Any]) → SchemaScheduleSettings

Produces an instance of the SchemaScheduleSettings from decoded data.

Parameters:: data (Mapping[str, Any]) – The JSON decoded data.
Returns:: An instance of the SchemaScheduleSettings.
Return type:: SchemaScheduleSettings

hour_of_day: int

model_config: ClassVar[ConfigDict] = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

repeats: RepeatSetting

shard_filter: ShardFilter | None

should_run(hour_of_day: int, day_of_week: int) → bool

Whether the schema should be run.

Parameters:

hour_of_day (int) – The hour of day to check.
day_of_week (int) – The day of week to check.

Returns:

Whether the schema should run.

Return type:

bool