beatdrop.entries package¶

Submodules¶

beatdrop.entries.crontab_entry module¶

class beatdrop.entries.crontab_entry.CrontabEntry

Bases: ScheduleEntry

Crontab style schedule entry based on UTC time.

Parameters:

key (str) – A unique key for the schedule entry.
enabled (bool) – Enable this entry to be scheduled.
task (str) – The full python path to the task to run.
args (Optional[Tuple[Any, ...]]) – Positional arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
kwargs (Optional[Dict[str, Any]]) – Keyword arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
cron_expression (str) – Crontab style date and time expression. croniter package is currently used as the parser. https://pypi.org/project/croniter/
last_sent_at (datetime.datetime, optional) – Client read only field Last time the entry was sent. Naive datetime in UTC.

_logger

Logger.

Type:: ClassVar

client_read_only_fields

Client read only list of fields. Enumerates the fields that are not normally saved when a client wants to save the entry. This is done because the client manages these fields. So they are are updated when the scheduler runs them.

Type:: ClassVar[List[str]] = []

class Config

Bases: object

validate_assignment = True

__iter__() → TupleGenerator: so dict(model) works

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]: Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str: Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult: Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None: Same as update_forward_refs but will not raise exception when forward references are not defined.

args: Tuple[Any, ...] | None

client_read_only_fields: ClassVar[List[str]] = ['last_sent_at']

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

cron_expression: str

dict(*args, **kwargs) → dict

Override of the pydantic dict method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: Dictionary representation of the model.
Return type:: dict

due_in() → timedelta

Returns when the schedule entry should run next.

Returns:: Time left until this entry should be scheduled. Zero or negative timedeltas mean it should be run.
Return type:: datetime.timedelta
Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

enabled: bool

classmethod from_orm(obj: Any) → Model

json(*args, **kwargs) → str

Override of the pydantic json method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: JSON representation of the model.
Return type:: str

key: str

kwargs: Dict[str, Any] | None

last_sent_at: datetime

classmethod parse_file(path: str | Path, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod parse_obj(obj: Any) → Model

classmethod parse_raw(b: str | bytes, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') → DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) → str

sent()

Called when the entry has been sent for execution.

This should be used to update any metadata as necessary for the entry. Like the last sent time etc.

Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

task: str

classmethod update_forward_refs(**localns: Any) → None: Try to update ForwardRefs on fields based on this Model, globalns and localns.

classmethod validate(value: Any) → Model

beatdrop.entries.crontab_tz_entry module¶

class beatdrop.entries.crontab_tz_entry.CrontabTZEntry

Bases: ScheduleEntry

Crontab style schedule entry based on a specific timezone.

Useful if you want a schedule entry to be sent during the same time in DST timezones.

For example, we want an entry to be sent at 1pm US/Eastern. With the normal CrontabEntry, we could set the cron_expression to "0 17 * * *" in the summer, but then in the winter the entry is now sent at 2pm US/Eastern instead.

The CrontabTZEntry solves this by setting the cron_expression to "0 13 * * *" and the timezone to "US/Eastern". The schedule entry is due at 1pm, US/Eastern every day.

Parameters:

key (str) – A unique key for the schedule entry.
enabled (bool) – Enable this entry to be scheduled.
task (str) – The full python path to the task to run.
args (Optional[Tuple[Any, ...]]) – Positional arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
kwargs (Optional[Dict[str, Any]]) – Keyword arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
cron_expression (str) – Crontab style date and time expression. croniter package is currently used as the parser. https://pypi.org/project/croniter/
timezone (str) – The timezone string. pytz library is used to parse this and create aware datetimes.
last_sent_at (datetime.datetime, optional) – Client read only field Last time the entry was sent. Naive datetime in UTC.

_logger

Logger.

Type:: ClassVar

client_read_only_fields

Client read only list of fields. Enumerates the fields that are not normally saved when a client wants to save the entry. This is done because the client manages these fields. So they are are updated when the scheduler runs them.

Type:: ClassVar[List[str]] = []

class Config

Bases: object

validate_assignment = True

__iter__() → TupleGenerator: so dict(model) works

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]: Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str: Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult: Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None: Same as update_forward_refs but will not raise exception when forward references are not defined.

args: Tuple[Any, ...] | None

client_read_only_fields: ClassVar[List[str]] = ['last_sent_at']

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

cron_expression: str

dict(*args, **kwargs) → dict

Override of the pydantic dict method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: Dictionary representation of the model.
Return type:: dict

due_in() → timedelta

Returns when the schedule entry should run next.

Returns:: Time left until this entry should be scheduled. Zero or negative timedeltas mean it should be run.
Return type:: datetime.timedelta
Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

enabled: bool

classmethod from_orm(obj: Any) → Model

json(*args, **kwargs) → str

Override of the pydantic json method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: JSON representation of the model.
Return type:: str

key: str

kwargs: Dict[str, Any] | None

last_sent_at: datetime

classmethod parse_file(path: str | Path, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod parse_obj(obj: Any) → Model

classmethod parse_raw(b: str | bytes, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') → DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) → str

sent()

Called when the entry has been sent for execution.

This should be used to update any metadata as necessary for the entry. Like the last sent time etc.

Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

task: str

timezone: str

classmethod update_forward_refs(**localns: Any) → None: Try to update ForwardRefs on fields based on this Model, globalns and localns.

classmethod valid_timezone(v)

classmethod validate(value: Any) → Model

beatdrop.entries.event_entry module¶

class beatdrop.entries.event_entry.EventEntry

Bases: ScheduleEntry

Event schedules are for a one time event.

After they run once they are disabled. sched.enabled = False

Parameters:

key (str) – A unique key for the schedule entry.
enabled (bool) – Enable this entry to be scheduled.
task (str) – The full python path to the task to run.
args (Optional[Tuple[Any, ...]]) – Positional arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
kwargs (Optional[Dict[str, Any]]) – Keyword arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
due_at (datetime.datetime) – The due at datetime. Takes naive or aware datetimes. Naive datetimes are assumed to be in UTC.

_logger

Logger.

Type:: ClassVar

client_read_only_fields

Client read only list of fields. Enumerates the fields that are not normally saved when a client wants to save the entry. This is done because the client manages these fields. So they are are updated when the scheduler runs them.

Type:: ClassVar[List[str]], optional

class Config

Bases: object

validate_assignment = True

__iter__() → TupleGenerator: so dict(model) works

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]: Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str: Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult: Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None: Same as update_forward_refs but will not raise exception when forward references are not defined.

args: Tuple[Any, ...] | None

client_read_only_fields: ClassVar[List[str]] = ['was_sent']

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

dict(*args, **kwargs) → dict

Override of the pydantic dict method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: Dictionary representation of the model.
Return type:: dict

due_at: datetime

due_in() → timedelta

Returns when the schedule entry should run next.

Returns:: Time left until this entry should be scheduled. Zero or negative timedeltas mean it should be run.
Return type:: datetime.timedelta
Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

enabled: bool

classmethod from_orm(obj: Any) → Model

json(*args, **kwargs) → str

Override of the pydantic json method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: JSON representation of the model.
Return type:: str

key: str

kwargs: Dict[str, Any] | None

classmethod parse_file(path: str | Path, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod parse_obj(obj: Any) → Model

classmethod parse_raw(b: str | bytes, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') → DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) → str

sent()

Called when the entry has been sent for execution.

This should be used to update any metadata as necessary for the entry. Like the last sent time etc.

Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

task: str

classmethod update_forward_refs(**localns: Any) → None: Try to update ForwardRefs on fields based on this Model, globalns and localns.

classmethod validate(value: Any) → Model

was_sent: bool

beatdrop.entries.interval_entry module¶

class beatdrop.entries.interval_entry.IntervalEntry

Bases: ScheduleEntry

Interval schedule entries are sent every period amount of time.

Parameters:

key (str) – A unique key for the schedule entry.
enabled (bool) – Enable this entry to be scheduled.
task (str) – The full python path to the task to run.
args (Optional[Tuple[Any, ...]]) – Positional arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
kwargs (Optional[Dict[str, Any]]) – Keyword arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
period (datetime.timedelta) – How often to run the schedule entry.
last_sent_at (datetime.datetime, optional) – Client read only field Last time the entry was sent

_logger

Logger.

Type:: ClassVar

client_read_only_fields

Client read only list of fields. Enumerates the fields that are not normally saved when a client wants to save the entry. This is done because the client manages these fields. So they are are updated when the scheduler runs them.

Type:: ClassVar[List[str]] = []

class Config

Bases: object

validate_assignment = True

__iter__() → TupleGenerator: so dict(model) works

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]: Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str: Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult: Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None: Same as update_forward_refs but will not raise exception when forward references are not defined.

args: Tuple[Any, ...] | None

client_read_only_fields: ClassVar[List[str]] = ['last_sent_at']

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

dict(*args, **kwargs) → dict

Override of the pydantic dict method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: Dictionary representation of the model.
Return type:: dict

due_in() → timedelta

Returns when the schedule entry should run next.

Returns:: Time left until this entry should be scheduled. Zero or negative timedeltas mean it should be run.
Return type:: datetime.timedelta
Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

enabled: bool

classmethod from_orm(obj: Any) → Model

json(*args, **kwargs) → str

Override of the pydantic json method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: JSON representation of the model.
Return type:: str

key: str

kwargs: Dict[str, Any] | None

last_sent_at: datetime

classmethod parse_file(path: str | Path, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod parse_obj(obj: Any) → Model

classmethod parse_raw(b: str | bytes, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

period: timedelta

classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') → DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) → str

sent()

Called when the entry has been sent for execution.

This should be used to update any metadata as necessary for the entry. Like the last sent time etc.

Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

task: str

classmethod timedelta_positive(v: timedelta) → timedelta

classmethod update_forward_refs(**localns: Any) → None: Try to update ForwardRefs on fields based on this Model, globalns and localns.

classmethod validate(value: Any) → Model

beatdrop.entries.schedule_entry module¶

class beatdrop.entries.schedule_entry.ScheduleEntry

Bases: BaseModel

Base ScheduleEntry.

All Schedule entries must implement the methods:

due_in - returns timedelta until it should be run again.
sent - called by the scheduler to let the entry know its task was sent for execution.

See their docstrings for more details.

A basic __str__ method is also included. It’s recommended to customize it in subclasses for better logging.

Parameters:

key (str) – A unique key for the schedule entry.
enabled (bool) – Enable this entry to be scheduled.
task (str) – The full python path to the task to run.
args (Optional[Tuple[Any, ...]]) – Positional arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.
kwargs (Optional[Dict[str, Any]]) – Keyword arguments to pass the task. These will be serialized/deserialized as JSON. jsonpickle is used to serialize and deserialize these.

_logger

Logger.

Type:: ClassVar

client_read_only_fields

Client read only list of fields. Enumerates the fields that are not normally saved when a client wants to save the entry. This is done because the client manages these fields. So they are are updated when the scheduler runs them.

Type:: ClassVar[List[str]] = []

class Config

Bases: object

validate_assignment = True

__iter__() → TupleGenerator: so dict(model) works

__pretty__(fmt: Callable[[Any], Any], **kwargs: Any) → Generator[Any, None, None]: Used by devtools (https://python-devtools.helpmanual.io/) to provide a human readable representations of objects

__repr_name__() → str: Name of the instance’s class, used in __repr__.

__rich_repr__() → RichReprResult: Get fields for Rich library

classmethod __try_update_forward_refs__(**localns: Any) → None: Same as update_forward_refs but will not raise exception when forward references are not defined.

args: Tuple[Any, ...] | None

client_read_only_fields: ClassVar[List[str]] = []

classmethod construct(_fields_set: SetStr | None = None, **values: Any) → Model: Creates a new model setting __dict__ and __fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed. Behaves as if Config.extra = ‘allow’ was set since it adds all passed values

Duplicate a model, optionally choose which fields to include, exclude and change.

Parameters:

include – fields to include in new model
exclude – fields to exclude from new model, as with values this takes precedence over include
update – values to change/add in the new model. Note: the data is not validated before creating the new model: you should trust this data
deep – set to True to make a deep copy of the model

Returns:

new model instance

dict(*args, **kwargs) → dict

Override of the pydantic dict method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: Dictionary representation of the model.
Return type:: dict

due_in() → timedelta

Returns when the schedule entry should run next.

Returns:: Time left until this entry should be scheduled. Zero or negative timedeltas mean it should be run.
Return type:: datetime.timedelta
Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

enabled: bool

classmethod from_orm(obj: Any) → Model

json(*args, **kwargs) → str

Override of the pydantic json method.

Adds the __beatdrop_type__ field for the SchedulerEntry Class name.

Returns:: JSON representation of the model.
Return type:: str

key: str

kwargs: Dict[str, Any] | None

classmethod parse_file(path: str | Path, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod parse_obj(obj: Any) → Model

classmethod parse_raw(b: str | bytes, *, content_type: str = None, encoding: str = 'utf8', proto: Protocol = None, allow_pickle: bool = False) → Model

classmethod schema(by_alias: bool = True, ref_template: str = '#/definitions/{model}') → DictStrAny

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/definitions/{model}', **dumps_kwargs: Any) → str

sent() → None

Called when the entry has been sent for execution.

This should be used to update any metadata as necessary for the entry. Like the last sent time etc.

Raises:: beatdrop.exceptions.MethodNotImplementedError – Must implement this method in subclass.

task: str

classmethod update_forward_refs(**localns: Any) → None: Try to update ForwardRefs on fields based on this Model, globalns and localns.

classmethod validate(value: Any) → Model