encommon.times package#

Subpackages#

Submodules#

encommon.times.common module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

encommon.times.duration module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.duration.Duration(seconds: int | float, smart: bool = True, groups: int = 7)[source]#

Bases: object

Convert the provided seconds in a human friendly format.

Example: Common Methods#

>>> durate = Duration(6048e4)
>>> durate.short
'1y 11mon 5d'
>>> durate.compact
'1y11mon5d'
>>> durate.verbose
'1 year, 11 months, 5 days'

Example: Dump the Units#

>>> durate = Duration(6048e4)
>>> durate.units()
{'year': 1, 'month': 11, 'day': 5}

Example: Disable Smart Seconds#

>>> durate = Duration(7201, False)
>>> durate.compact
'2h1s'
>>> durate.verbose
'2 hours, 1 second'

Example: Basic Operators#

>>> durate1 = Duration(1000)
>>> durate2 = Duration(2000)
>>> durate1 == durate2
False
>>> durate1 != durate2
True
>>> durate1 > durate2
False
>>> durate1 >= durate2
False
>>> durate1 < durate2
True
>>> durate1 <= durate2
True
>>> durate1 + durate2
3000.0
>>> durate1 - durate2
-1000.0
>>> durate2 - durate1
1000.0
param seconds:

Period in seconds that will be iterated.

param smart:

Determines if we hide seconds after minute.

param groups:

Determine the quantity of groups to show, ensuring larger units are returned before smaller.

property compact: str#

Return the compact format determined by source duration.

Returns:

Compact format determined by source duration.

property groups: int#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property short: str#

Return the compact format determined by source duration.

Returns:

Compact format determined by source duration.

property smart: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property source: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

units(short: bool = False) → dict[Literal['year', 'month', 'week', 'day', 'hour', 'minute', 'second', 'y', 'mon', 'w', 'd', 'h', 'm', 's'], int][source]#

Return the groups of time units with each relevant value.

Parameters:

short – Determine if we should use the short hand.

Returns:

Groups of time units with each relevant value.

property verbose: str#

Return the verbose format determined by source duration.

Returns:

Verbose format determined by source duration.

encommon.times.params module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

pydantic model encommon.times.params.TimerParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "TimerParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "timer": {
         "description": "Seconds used for the interval",
         "title": "Timer",
         "type": "number"
      },
      "start": {
         "description": "Optional value of timer start",
         "minLength": 1,
         "title": "Start",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "timer",
      "start"
   ]
}

Config:

  • extra: str = forbid

Fields:
field start: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Optional value of timer start', metadata=[MinLen(min_length=1)])] [Required]#

Optional value of timer start

Constraints:

  • min_length = 1

field timer: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Seconds used for the interval')] [Required]#

Seconds used for the interval

pydantic model encommon.times.params.TimersParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "TimersParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "timers": {
         "additionalProperties": {
            "$ref": "#/$defs/TimerParams"
         },
         "description": "Seconds used for the interval",
         "minProperties": 0,
         "title": "Timers",
         "type": "object"
      }
   },
   "$defs": {
      "TimerParams": {
         "additionalProperties": false,
         "description": "Process and validate the core configuration parameters.",
         "properties": {
            "timer": {
               "description": "Seconds used for the interval",
               "title": "Timer",
               "type": "number"
            },
            "start": {
               "description": "Optional value of timer start",
               "minLength": 1,
               "title": "Start",
               "type": "string"
            }
         },
         "required": [
            "timer",
            "start"
         ],
         "title": "TimerParams",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "timers"
   ]
}

Config:

  • extra: str = forbid

Fields:
field timers: Annotated[dict[str, TimerParams], FieldInfo(annotation=NoneType, required=True, description='Seconds used for the interval', metadata=[MinLen(min_length=0)])] [Required]#

Seconds used for the interval

Constraints:

  • min_length = 0

pydantic model encommon.times.params.WindowParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "WindowParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "window": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "additionalProperties": {
                  "type": "integer"
               },
               "type": "object"
            }
         ],
         "description": "Period for scheduling window",
         "minLength": 1,
         "title": "Window"
      },
      "start": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Determine when scope begins",
         "title": "Start"
      },
      "stop": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Determine when scope ends",
         "title": "Stop"
      },
      "anchor": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Optional anchor of the window",
         "title": "Anchor"
      },
      "delay": {
         "description": "Time period of schedule delay",
         "minimum": 0,
         "title": "Delay",
         "type": "number"
      }
   },
   "additionalProperties": false,
   "required": [
      "window",
      "delay"
   ]
}

Config:

  • extra: str = forbid

Fields:
field anchor: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional anchor of the window', metadata=[MinLen(min_length=1)])] = None#

Optional anchor of the window

Constraints:

  • min_length = 1

field delay: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Time period of schedule delay', metadata=[Ge(ge=0)])] [Required]#

Time period of schedule delay

Constraints:

  • ge = 0

field start: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Determine when scope begins', metadata=[MinLen(min_length=1)])] = None#

Determine when scope begins

Constraints:

  • min_length = 1

field stop: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Determine when scope ends', metadata=[MinLen(min_length=1)])] = None#

Determine when scope ends

Constraints:

  • min_length = 1

field window: Annotated[str | dict[str, int], FieldInfo(annotation=NoneType, required=True, description='Period for scheduling window', metadata=[MinLen(min_length=1)])] [Required]#

Period for scheduling window

Constraints:

  • min_length = 1

pydantic model encommon.times.params.WindowsParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "WindowsParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "windows": {
         "additionalProperties": {
            "$ref": "#/$defs/WindowParams"
         },
         "description": "Period for scheduling windows",
         "minProperties": 0,
         "title": "Windows",
         "type": "object"
      }
   },
   "$defs": {
      "WindowParams": {
         "additionalProperties": false,
         "description": "Process and validate the core configuration parameters.",
         "properties": {
            "window": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "additionalProperties": {
                        "type": "integer"
                     },
                     "type": "object"
                  }
               ],
               "description": "Period for scheduling window",
               "minLength": 1,
               "title": "Window"
            },
            "start": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Determine when scope begins",
               "title": "Start"
            },
            "stop": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Determine when scope ends",
               "title": "Stop"
            },
            "anchor": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Optional anchor of the window",
               "title": "Anchor"
            },
            "delay": {
               "description": "Time period of schedule delay",
               "minimum": 0,
               "title": "Delay",
               "type": "number"
            }
         },
         "required": [
            "window",
            "delay"
         ],
         "title": "WindowParams",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "windows"
   ]
}

Config:

  • extra: str = forbid

Fields:
field windows: Annotated[dict[str, WindowParams], FieldInfo(annotation=NoneType, required=True, description='Period for scheduling windows', metadata=[MinLen(min_length=0)])] [Required]#

Period for scheduling windows

Constraints:

  • min_length = 0

encommon.times.parse module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

encommon.times.parse.parse_time(source: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, format: str | None = None, tzname: str | None = None) → datetime[source]#

Parse provided time value with various supported formats.

Example

Description

‘now’

Uses the current time

None, ‘None’, ‘null’

Uses the current time

‘min’, float(‘-inf’)

Uses the minimum time

‘max’, float(‘inf’)

Uses the maximum time

object

Provide datetime or Time

2000-01-01T00:00:00Z

Uses strptime and dateutil

1645843626.2

Provide Unix epoch

‘-1h@h+5m’

Provide snaptime syntax

Example#

>>> parse_time('1/1/2000 12:00am')
datetime.datetime(2000, 1, 1, 0...

Example#

>>> parse_time(0)
datetime.datetime(1970, 1, 1, 0...
param source:

Time in various forms that will be parsed.

param anchor:

Optional relative time; for snap notation.

param format:

Optional format when source is timestamp.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

returns:

Python datetime object containing related time.

encommon.times.parse.shift_time(notate: str, anchor: str | int | float | datetime | Time | None = None, *, tzname: str | None = None) → datetime[source]#

Shift provided time value using snaptime relative syntax.

Example#

>>> shift_time('-1h', anchor='1/1/2000 12:00am')
datetime.datetime(1999, 12, 31, 23...

Example#

>>> shift_time('-1h@d', anchor='1/1/2000 12:00am')
datetime.datetime(1999, 12, 31, 0...
param notate:

Syntax compatable using snaptime library.

param anchor:

Optional relative time; for snap notation.

param tzname:

Name of the timezone associated to anchor. This is not relevant in timezone included in anchor.

returns:

Python datetime object containing related time.

encommon.times.parse.since_time(start: str | int | float | datetime | Time, stop: str | int | float | datetime | Time | None = None) → float[source]#

Determine the time in seconds occurring between values.

Example#

>>> start = parse_time(0)
>>> stop = parse_time(1)
>>> since_time(start, stop)
1.0
param start:

Time in various forms that will be parsed.

param stop:

Time in various forms that will be parsed.

returns:

Time in seconds occurring between the values.

encommon.times.parse.string_time(source: str, *, formats: str | list[str] | None = None, tzname: str | None = None) → datetime[source]#

Parse provided time value with various supported formats.

Example#

>>> string_time('2000-01-01T00:00:00Z')
datetime.datetime(2000, 1, 1, 0...

Example#

>>> string_time('2000/01/01')
datetime.datetime(2000, 1, 1, 0...
param source:

Time in various forms that will be parsed.

param formats:

Various formats compatable with strptime.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

returns:

Python datetime object containing related time.

encommon.times.time module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.time.Time(source: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, format: str = '%Y-%m-%dT%H:%M:%S.%f%z', tzname: str | None = None)[source]#

Bases: object

Interact with various time functions through one wrapper.

Example#

>>> time = Time('1/1/2000 12:00am')
>>> time.stamp()
'2000-01-01T00:00:00.000000+0000'
>>> time.stamp('%m/%d/%Y')
'01/01/2000'

Example#

>>> time.epoch
946684800.0
>>> time.time
datetime.time(0, 0)
>>> time.simple
'2000-01-01T00:00:00+0000'
>>> time.human
'01/01/2000 12:00AM UTC'

Example#

>>> time.before
Time('1999-12-31T23:59:59.999999+0000')
>>> time.after
Time('2000-01-01T00:00:00.000001+0000')

Example#

>>> time.shift('-1d')
Time('1999-12-31T00:00:00.000000+0000')

Example#

>>> time.shifz('US/Central')
Time('1999-12-31T18:00:00.000000-0600')

Example#

>>> time = Time('-1s')
>>> int(time.since)
1

Example#

>>> time1 = Time('1/1/2000 12:00am')
>>> time2 = Time('1/1/2000 12:00am')
>>> time1 - time2
0.0
>>> time1 + time2
1893369600.0
param source:

Time in various forms that will be parsed.

param anchor:

Optional relative time; for snap notation.

param format:

Optional format when source is timestamp.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

property after: Time#

Return new object containing time just after the time.

Returns:

Object containing time just after the time.

property before: Time#

Return new object containing time just before the time.

Returns:

Object containing time just before the time.

property elapsed: float#

Determine the time in seconds occurring since instance.

Returns:

Time in seconds occurring since the instance.

property epoch: float#

Return the seconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

property human: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property mpoch: float#

Return milliseconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

shift(notate: str) → Time[source]#

Return the new instance of object shifted using snaptime.

Parameters:

notate – Syntax compatable using snaptime library.

Returns:

New instance of the class using shifted time.

shifz(tzname: str) → Time[source]#

Return the new instance of object shifted using datetime.

Parameters:

tzname – Name of the timezone associated to source. This is not relevant in timezone included in source.

Returns:

New instance of the class using shifted time.

property simple: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property since: float#

Determine the time in seconds occurring since instance.

Returns:

Time in seconds occurring since the instance.

property source: datetime#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property spoch: int#

Return the seconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

stamp(format: str = '%Y-%m-%dT%H:%M:%S.%f%z', tzname: str | None = None) → str[source]#

Return the timestamp using provided format for instance.

Parameters:

  • format – Optional format when source is timestamp.

  • tzname – Name of the timezone associated to source. This is not relevant in timezone included in source.

Returns:

Timestamp using provided format for instance.

property subsec: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property time: time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

encommon.times.timer module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.timer.Timer(timer: int | float, *, start: str | int | float | datetime | Time | None = None)[source]#

Bases: object

Process and operate the timer using provided interval.

Example#

>>> timer = Timer(1)
>>> timer.ready()
False
>>> sleep(1)
>>> timer.ready()
True
param seconds:

Period of time which must pass in timer.

param started:

Override default of using current time.

pause(update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

property remains: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property since: float#

Return the seconds that have elapsed since the interval.

Returns:

Seconds that have elapsed since the interval.

property time: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property timer: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(value: str | int | float | datetime | Time | None = None) → None[source]#

Update the timer from the provided parasable time value.

Parameters:

value – Override the time updated for timer value.

encommon.times.timers module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.timers.Timers(params: TimersParams | None = None, *, store: str = 'sqlite:///:memory:', table: str = 'timers', group: str = 'default')[source]#

Bases: object

Track timers on unique key determining when to proceed.

Warning

This class will use an in-memory database for cache, unless a cache file is explicity defined.

Example#

>>> source = {'one': TimerParams(timer=1)}
>>> params = TimersParams(timers=source)
>>> timers = Timers(params)
>>> timers.ready('one')
False
>>> sleep(1)
>>> timers.ready('one')
True
param params:

Parameters used to instantiate the class.

param store:

Optional database path for keeping state.

param group:

Optional override for default group name.

property children: dict[str, Timer]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

create(unique: str, params: TimerParams) → Timer[source]#

Create a new timer using the provided input parameters.

Parameters:

  • unique – Unique identifier for the related child.

  • params – Parameters used to instantiate the class.

Returns:

Newly constructed instance of related class.

delete(unique: str) → None[source]#

Delete the timer from the internal dictionary reference.

Note

This is a graceful method, will not raise exception when the provided unique value does not exist.

Parameters:

unique – Unique identifier for the related child.

property group: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: TimersParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

pause(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

save_children() → None[source]#

Save the child caches from the attribute into database.

select(unique: str) → TimersRecord | None[source]#

Return the record from within the table in the database.

Returns:

Record from within the table in the database.

property store: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_engine: Engine#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_session: Session#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_table: Type[TimersTable]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(unique: str, value: str | int | float | datetime | Time | None = None) → None[source]#

Update the timer from the provided parasable time value.

Parameters:

  • unique – Unique identifier for the related child.

  • value – Override the time updated for timer value.

pydantic model encommon.times.timers.TimersRecord[source]#

Bases: BaseModel

Contain the information of the record from the database.

Parameters:

record – Record from the SQLAlchemy query routine.

Show JSON schema
{
   "title": "TimersRecord",
   "description": "Contain the information of the record from the database.\n\n:param record: Record from the SQLAlchemy query routine.",
   "type": "object",
   "properties": {
      "group": {
         "description": "Group the children by category",
         "minLength": 1,
         "title": "Group",
         "type": "string"
      },
      "unique": {
         "description": "Unique identifier for the child",
         "minLength": 1,
         "title": "Unique",
         "type": "string"
      },
      "last": {
         "description": "Last interval for the period",
         "maxLength": 32,
         "minLength": 20,
         "title": "Last",
         "type": "string"
      },
      "update": {
         "description": "When the record was updated",
         "maxLength": 32,
         "minLength": 20,
         "title": "Update",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "group",
      "unique",
      "last",
      "update"
   ]
}

Config:

  • extra: str = forbid

Fields:
field group: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Group the children by category', metadata=[MinLen(min_length=1)])] [Required]#

Group the children by category

Constraints:

  • min_length = 1

field last: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Last interval for the period', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])] [Required]#

Last interval for the period

Constraints:

  • min_length = 20

  • max_length = 32

field unique: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Unique identifier for the child', metadata=[MinLen(min_length=1)])] [Required]#

Unique identifier for the child

Constraints:

  • min_length = 1

field update: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='When the record was updated', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])] [Required]#

When the record was updated

Constraints:

  • min_length = 20

  • max_length = 32

class encommon.times.timers.TimersTable(**kwargs: Any)[source]#

Bases: DeclarativeBase

Schematic for the database operations using SQLAlchemy.

Note

Fields are not completely documented for this model.

group = Column(None, String(), table=None, primary_key=True, nullable=False)#
last = Column(None, String(), table=None, nullable=False)#
metadata: ClassVar[MetaData] = MetaData()#

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>#

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

unique = Column(None, String(), table=None, primary_key=True, nullable=False)#
update = Column(None, String(), table=None, nullable=False)#

encommon.times.unitime module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

encommon.times.unitime.unitime(input: int | float | str) → int[source]#

Return the seconds in integer format for provided input.

Example#

>>> unitime('1d')
86400
>>> unitime('1y')
31536000
>>> unitime('1w3d4h')
878400
param input:

Input that will be converted into seconds.

returns:

Seconds in integer format for provided input.

encommon.times.utils module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

encommon.times.utils.findtz(tzname: str | None = None) → tzinfo[source]#

Return the located timezone object for the provided name.

Example#

>>> findtz('US/Central')
tzfile('/usr/share/zoneinfo/US/Central')
param tzname:

Name of the timezone associated to source.

returns:

Located timezone object for the provided name.

encommon.times.utils.strftime(source: datetime, format: str) → str[source]#

Return the timestamp string for datetime object provided.

Note

This function is extremely pedantic and cosmetic.

Example#

>>> dtime = datetime(2023, 1, 1)
>>> strftime(dtime, '%Y')
'2023'
param source:

Python datetime instance containing source.

param format:

Format for the timestamp string returned.

returns:

Timestamp string for datetime object provided.

encommon.times.utils.strptime(source: str, formats: str | list[str]) → datetime[source]#

Parse provided time value with various supported formats.

Example#

>>> strptime('2023', '%Y')
datetime.datetime(2023, 1, 1, 0...
param source:

Time in various forms that will be parsed.

param formats:

Various formats compatable with strptime.

returns:

Python datetime object containing related time.

encommon.times.utils.utcdatetime(*args: Any, **kwargs: Any) → datetime[source]#

Return the instance of datetime within the UTC timezone.

Warning

If no arguments are provided, returns current time.

Example#

>>> utcdatetime(1970, 1, 1)
datetime.datetime(1970, 1, 1, 0...
param args:

Positional arguments passed for downstream.

param kwargs:

Keyword arguments passed for downstream.

returns:

Instance of datetime within the UTC timezone.

encommon.times.window module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.window.Window(window: str | dict[str, int], start: str | int | float | datetime | Time | None = None, stop: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, delay: float = 0)[source]#

Bases: object

Process and operate crontab or interval based schedule.

Example#

>>> window = Window('* * * * *', '-4m@m')
>>> [window.ready() for _ in range(6)]
[True, True, True, True, True, False]
param window:

Parameters for defining scheduled time.

param start:

Determine the start for scheduling window.

param stop:

Determine the ending for scheduling window.

param anchor:

Optionally define time anchor for window.

param delay:

Period of time schedulng will be delayed.

property anchor: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property delay: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property last: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property latest: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property next: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

ready(update: bool = True) → bool[source]#

Walk the internal time using current position in schedule.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating outcome from the operation.

property soonest: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property start: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property stop: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(value: str | int | float | datetime | Time | None = None) → None[source]#

Update the window from the provided parasable time value.

Parameters:

value – Override the time updated for window value.

property walked: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property window: str | dict[str, int]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

encommon.times.window.window_croniter(schedule: str, anchor: str | int | float | datetime | Time, backward: bool = False) → tuple[Time, Time][source]#

Determine next and previous times for cronjob schedule.

Parameters:

  • window – Parameters for defining scheduled time.

  • anchor – Optionally define time anchor for window.

  • backward – Optionally operate the window backward.

Returns:

Next and previous windows for schedule window.

encommon.times.window.window_interval(schedule: dict[str, int], anchor: str | int | float | datetime | Time, backward: bool = False) → tuple[Time, Time][source]#

Determine next and previous times for interval schedule.

Parameters:

  • window – Parameters for defining scheduled time.

  • anchor – Optionally define time anchor for window.

  • backward – Optionally operate the window backward.

Returns:

Next and previous windows for schedule window.

encommon.times.windows module#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.windows.Windows(params: WindowsParams | None = None, start: str | int | float | datetime | Time = 'now', stop: str | int | float | datetime | Time = '3000-01-01', *, store: str = 'sqlite:///:memory:', table: str = 'windows', group: str = 'default')[source]#

Bases: object

Track windows on unique key determining when to proceed.

Warning

This class will use an in-memory database for cache, unless a cache file is explicity defined.

Example#

>>> source = {'one': WindowParams(window=1)}
>>> params = WindowsParams(windows=source)
>>> windows = Windows(params, '-2s', 'now')
>>> [windows.ready('one') for x in range(4)]
[True, True, True, False]
param params:

Parameters used to instantiate the class.

param start:

Determine the start for scheduling window.

param stop:

Determine the ending for scheduling window.

param store:

Optional database path for keeping state.

param group:

Optional override for default group name.

property children: dict[str, Window]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

create(unique: str, params: WindowParams) → Window[source]#

Create a new window using the provided input parameters.

Parameters:

  • unique – Unique identifier for the related child.

  • params – Parameters used to instantiate the class.

Returns:

Newly constructed instance of related class.

delete(unique: str) → None[source]#

Delete the window from the internal dictionary reference.

Note

This is a graceful method, will not raise exception when the provided unique value does not exist.

Parameters:

unique – Unique identifier for the related child.

property group: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: WindowsParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

pause(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

save_children() → None[source]#

Save the child caches from the attribute into database.

select(unique: str) → WindowsRecord | None[source]#

Return the record from within the table in the database.

Returns:

Record from within the table in the database.

property start: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property stop: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_engine: Engine#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_session: Session#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_table: Type[WindowsTable]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(unique: str, value: str | int | float | datetime | Time | None = None) → None[source]#

Update the window from the provided parasable time value.

Parameters:

  • unique – Unique identifier for the related child.

  • value – Override the time updated for window value.

pydantic model encommon.times.windows.WindowsRecord[source]#

Bases: BaseModel

Contain the information of the record from the database.

Parameters:

record – Record from the SQLAlchemy query routine.

Show JSON schema
{
   "title": "WindowsRecord",
   "description": "Contain the information of the record from the database.\n\n:param record: Record from the SQLAlchemy query routine.",
   "type": "object",
   "properties": {
      "group": {
         "description": "Group the children by category",
         "minLength": 1,
         "title": "Group",
         "type": "string"
      },
      "unique": {
         "description": "Unique identifier for the child",
         "minLength": 1,
         "title": "Unique",
         "type": "string"
      },
      "last": {
         "description": "Last schedule for the period",
         "maxLength": 32,
         "minLength": 20,
         "title": "Last",
         "type": "string"
      },
      "next": {
         "description": "Next schedule for the period",
         "maxLength": 32,
         "minLength": 20,
         "title": "Next",
         "type": "string"
      },
      "update": {
         "description": "When the record was updated",
         "maxLength": 32,
         "minLength": 20,
         "title": "Update",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "group",
      "unique",
      "last",
      "next",
      "update"
   ]
}

Config:

  • extra: str = forbid

Fields:
field group: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Group the children by category', metadata=[MinLen(min_length=1)])] [Required]#

Group the children by category

Constraints:

  • min_length = 1

field last: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Last schedule for the period', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])] [Required]#

Last schedule for the period

Constraints:

  • min_length = 20

  • max_length = 32

field next: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Next schedule for the period', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])] [Required]#

Next schedule for the period

Constraints:

  • min_length = 20

  • max_length = 32

field unique: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Unique identifier for the child', metadata=[MinLen(min_length=1)])] [Required]#

Unique identifier for the child

Constraints:

  • min_length = 1

field update: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='When the record was updated', metadata=[MinLen(min_length=20), MaxLen(max_length=32)])] [Required]#

When the record was updated

Constraints:

  • min_length = 20

  • max_length = 32

class encommon.times.windows.WindowsTable(**kwargs: Any)[source]#

Bases: DeclarativeBase

Schematic for the database operations using SQLAlchemy.

Note

Fields are not completely documented for this model.

group = Column(None, String(), table=None, primary_key=True, nullable=False)#
last = Column(None, String(), table=None, nullable=False)#
metadata: ClassVar[MetaData] = MetaData()#

Refers to the _schema.MetaData collection that will be used for new _schema.Table objects.

See also

Accessing Table and Metadata

next = Column(None, String(), table=None, nullable=False)#
registry: ClassVar[_RegistryType] = <sqlalchemy.orm.decl_api.registry object>#

Refers to the _orm.registry in use where new _orm.Mapper objects will be associated.

unique = Column(None, String(), table=None, primary_key=True, nullable=False)#
update = Column(None, String(), table=None, nullable=False)#

Module contents#

Functions and routines associated with Enasis Network Common Library.

This file is part of Enasis Network software eco-system. Distribution is permitted, for more information consult the project license file.

class encommon.times.Duration(seconds: int | float, smart: bool = True, groups: int = 7)[source]#

Bases: object

Convert the provided seconds in a human friendly format.

Example: Common Methods#

>>> durate = Duration(6048e4)
>>> durate.short
'1y 11mon 5d'
>>> durate.compact
'1y11mon5d'
>>> durate.verbose
'1 year, 11 months, 5 days'

Example: Dump the Units#

>>> durate = Duration(6048e4)
>>> durate.units()
{'year': 1, 'month': 11, 'day': 5}

Example: Disable Smart Seconds#

>>> durate = Duration(7201, False)
>>> durate.compact
'2h1s'
>>> durate.verbose
'2 hours, 1 second'

Example: Basic Operators#

>>> durate1 = Duration(1000)
>>> durate2 = Duration(2000)
>>> durate1 == durate2
False
>>> durate1 != durate2
True
>>> durate1 > durate2
False
>>> durate1 >= durate2
False
>>> durate1 < durate2
True
>>> durate1 <= durate2
True
>>> durate1 + durate2
3000.0
>>> durate1 - durate2
-1000.0
>>> durate2 - durate1
1000.0
param seconds:

Period in seconds that will be iterated.

param smart:

Determines if we hide seconds after minute.

param groups:

Determine the quantity of groups to show, ensuring larger units are returned before smaller.

property compact: str#

Return the compact format determined by source duration.

Returns:

Compact format determined by source duration.

property groups: int#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property short: str#

Return the compact format determined by source duration.

Returns:

Compact format determined by source duration.

property smart: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property source: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

units(short: bool = False) → dict[Literal['year', 'month', 'week', 'day', 'hour', 'minute', 'second', 'y', 'mon', 'w', 'd', 'h', 'm', 's'], int][source]#

Return the groups of time units with each relevant value.

Parameters:

short – Determine if we should use the short hand.

Returns:

Groups of time units with each relevant value.

property verbose: str#

Return the verbose format determined by source duration.

Returns:

Verbose format determined by source duration.

class encommon.times.Time(source: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, format: str = '%Y-%m-%dT%H:%M:%S.%f%z', tzname: str | None = None)[source]#

Bases: object

Interact with various time functions through one wrapper.

Example#

>>> time = Time('1/1/2000 12:00am')
>>> time.stamp()
'2000-01-01T00:00:00.000000+0000'
>>> time.stamp('%m/%d/%Y')
'01/01/2000'

Example#

>>> time.epoch
946684800.0
>>> time.time
datetime.time(0, 0)
>>> time.simple
'2000-01-01T00:00:00+0000'
>>> time.human
'01/01/2000 12:00AM UTC'

Example#

>>> time.before
Time('1999-12-31T23:59:59.999999+0000')
>>> time.after
Time('2000-01-01T00:00:00.000001+0000')

Example#

>>> time.shift('-1d')
Time('1999-12-31T00:00:00.000000+0000')

Example#

>>> time.shifz('US/Central')
Time('1999-12-31T18:00:00.000000-0600')

Example#

>>> time = Time('-1s')
>>> int(time.since)
1

Example#

>>> time1 = Time('1/1/2000 12:00am')
>>> time2 = Time('1/1/2000 12:00am')
>>> time1 - time2
0.0
>>> time1 + time2
1893369600.0
param source:

Time in various forms that will be parsed.

param anchor:

Optional relative time; for snap notation.

param format:

Optional format when source is timestamp.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

property after: Time#

Return new object containing time just after the time.

Returns:

Object containing time just after the time.

property before: Time#

Return new object containing time just before the time.

Returns:

Object containing time just before the time.

property elapsed: float#

Determine the time in seconds occurring since instance.

Returns:

Time in seconds occurring since the instance.

property epoch: float#

Return the seconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

property human: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property mpoch: float#

Return milliseconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

shift(notate: str) → Time[source]#

Return the new instance of object shifted using snaptime.

Parameters:

notate – Syntax compatable using snaptime library.

Returns:

New instance of the class using shifted time.

shifz(tzname: str) → Time[source]#

Return the new instance of object shifted using datetime.

Parameters:

tzname – Name of the timezone associated to source. This is not relevant in timezone included in source.

Returns:

New instance of the class using shifted time.

property simple: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property since: float#

Determine the time in seconds occurring since instance.

Returns:

Time in seconds occurring since the instance.

property source: datetime#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property spoch: int#

Return the seconds since the Unix epoch for the instance.

Returns:

Seconds since the Unix epoch for the instance.

stamp(format: str = '%Y-%m-%dT%H:%M:%S.%f%z', tzname: str | None = None) → str[source]#

Return the timestamp using provided format for instance.

Parameters:

  • format – Optional format when source is timestamp.

  • tzname – Name of the timezone associated to source. This is not relevant in timezone included in source.

Returns:

Timestamp using provided format for instance.

property subsec: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property time: time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

class encommon.times.Timer(timer: int | float, *, start: str | int | float | datetime | Time | None = None)[source]#

Bases: object

Process and operate the timer using provided interval.

Example#

>>> timer = Timer(1)
>>> timer.ready()
False
>>> sleep(1)
>>> timer.ready()
True
param seconds:

Period of time which must pass in timer.

param started:

Override default of using current time.

pause(update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

property remains: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property since: float#

Return the seconds that have elapsed since the interval.

Returns:

Seconds that have elapsed since the interval.

property time: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property timer: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(value: str | int | float | datetime | Time | None = None) → None[source]#

Update the timer from the provided parasable time value.

Parameters:

value – Override the time updated for timer value.

pydantic model encommon.times.TimerParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "TimerParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "timer": {
         "description": "Seconds used for the interval",
         "title": "Timer",
         "type": "number"
      },
      "start": {
         "description": "Optional value of timer start",
         "minLength": 1,
         "title": "Start",
         "type": "string"
      }
   },
   "additionalProperties": false,
   "required": [
      "timer",
      "start"
   ]
}

Config:

  • extra: str = forbid

Fields:
field start: Annotated[str, FieldInfo(annotation=NoneType, required=True, description='Optional value of timer start', metadata=[MinLen(min_length=1)])] [Required]#

Optional value of timer start

Constraints:

  • min_length = 1

field timer: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Seconds used for the interval')] [Required]#

Seconds used for the interval

class encommon.times.Timers(params: TimersParams | None = None, *, store: str = 'sqlite:///:memory:', table: str = 'timers', group: str = 'default')[source]#

Bases: object

Track timers on unique key determining when to proceed.

Warning

This class will use an in-memory database for cache, unless a cache file is explicity defined.

Example#

>>> source = {'one': TimerParams(timer=1)}
>>> params = TimersParams(timers=source)
>>> timers = Timers(params)
>>> timers.ready('one')
False
>>> sleep(1)
>>> timers.ready('one')
True
param params:

Parameters used to instantiate the class.

param store:

Optional database path for keeping state.

param group:

Optional override for default group name.

property children: dict[str, Timer]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

create(unique: str, params: TimerParams) → Timer[source]#

Create a new timer using the provided input parameters.

Parameters:

  • unique – Unique identifier for the related child.

  • params – Parameters used to instantiate the class.

Returns:

Newly constructed instance of related class.

delete(unique: str) → None[source]#

Delete the timer from the internal dictionary reference.

Note

This is a graceful method, will not raise exception when the provided unique value does not exist.

Parameters:

unique – Unique identifier for the related child.

property group: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: TimersParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

pause(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

save_children() → None[source]#

Save the child caches from the attribute into database.

select(unique: str) → TimersRecord | None[source]#

Return the record from within the table in the database.

Returns:

Record from within the table in the database.

property store: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_engine: Engine#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_session: Session#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_table: Type[TimersTable]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(unique: str, value: str | int | float | datetime | Time | None = None) → None[source]#

Update the timer from the provided parasable time value.

Parameters:

  • unique – Unique identifier for the related child.

  • value – Override the time updated for timer value.

pydantic model encommon.times.TimersParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "TimersParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "timers": {
         "additionalProperties": {
            "$ref": "#/$defs/TimerParams"
         },
         "description": "Seconds used for the interval",
         "minProperties": 0,
         "title": "Timers",
         "type": "object"
      }
   },
   "$defs": {
      "TimerParams": {
         "additionalProperties": false,
         "description": "Process and validate the core configuration parameters.",
         "properties": {
            "timer": {
               "description": "Seconds used for the interval",
               "title": "Timer",
               "type": "number"
            },
            "start": {
               "description": "Optional value of timer start",
               "minLength": 1,
               "title": "Start",
               "type": "string"
            }
         },
         "required": [
            "timer",
            "start"
         ],
         "title": "TimerParams",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "timers"
   ]
}

Config:

  • extra: str = forbid

Fields:
field timers: Annotated[dict[str, TimerParams], FieldInfo(annotation=NoneType, required=True, description='Seconds used for the interval', metadata=[MinLen(min_length=0)])] [Required]#

Seconds used for the interval

Constraints:

  • min_length = 0

class encommon.times.Window(window: str | dict[str, int], start: str | int | float | datetime | Time | None = None, stop: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, delay: float = 0)[source]#

Bases: object

Process and operate crontab or interval based schedule.

Example#

>>> window = Window('* * * * *', '-4m@m')
>>> [window.ready() for _ in range(6)]
[True, True, True, True, True, False]
param window:

Parameters for defining scheduled time.

param start:

Determine the start for scheduling window.

param stop:

Determine the ending for scheduling window.

param anchor:

Optionally define time anchor for window.

param delay:

Period of time schedulng will be delayed.

property anchor: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property delay: float#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property last: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property latest: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property next: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

ready(update: bool = True) → bool[source]#

Walk the internal time using current position in schedule.

Parameters:

update – Determines whether or not time is updated.

Returns:

Boolean indicating outcome from the operation.

property soonest: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property start: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property stop: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(value: str | int | float | datetime | Time | None = None) → None[source]#

Update the window from the provided parasable time value.

Parameters:

value – Override the time updated for window value.

property walked: bool#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property window: str | dict[str, int]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

pydantic model encommon.times.WindowParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "WindowParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "window": {
         "anyOf": [
            {
               "type": "string"
            },
            {
               "additionalProperties": {
                  "type": "integer"
               },
               "type": "object"
            }
         ],
         "description": "Period for scheduling window",
         "minLength": 1,
         "title": "Window"
      },
      "start": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Determine when scope begins",
         "title": "Start"
      },
      "stop": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Determine when scope ends",
         "title": "Stop"
      },
      "anchor": {
         "anyOf": [
            {
               "minLength": 1,
               "type": "string"
            },
            {
               "type": "null"
            }
         ],
         "default": null,
         "description": "Optional anchor of the window",
         "title": "Anchor"
      },
      "delay": {
         "description": "Time period of schedule delay",
         "minimum": 0,
         "title": "Delay",
         "type": "number"
      }
   },
   "additionalProperties": false,
   "required": [
      "window",
      "delay"
   ]
}

Config:

  • extra: str = forbid

Fields:
field anchor: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Optional anchor of the window', metadata=[MinLen(min_length=1)])] = None#

Optional anchor of the window

Constraints:

  • min_length = 1

field delay: Annotated[float, FieldInfo(annotation=NoneType, required=True, description='Time period of schedule delay', metadata=[Ge(ge=0)])] [Required]#

Time period of schedule delay

Constraints:

  • ge = 0

field start: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Determine when scope begins', metadata=[MinLen(min_length=1)])] = None#

Determine when scope begins

Constraints:

  • min_length = 1

field stop: Annotated[str | None, FieldInfo(annotation=NoneType, required=False, default=None, description='Determine when scope ends', metadata=[MinLen(min_length=1)])] = None#

Determine when scope ends

Constraints:

  • min_length = 1

field window: Annotated[str | dict[str, int], FieldInfo(annotation=NoneType, required=True, description='Period for scheduling window', metadata=[MinLen(min_length=1)])] [Required]#

Period for scheduling window

Constraints:

  • min_length = 1

class encommon.times.Windows(params: WindowsParams | None = None, start: str | int | float | datetime | Time = 'now', stop: str | int | float | datetime | Time = '3000-01-01', *, store: str = 'sqlite:///:memory:', table: str = 'windows', group: str = 'default')[source]#

Bases: object

Track windows on unique key determining when to proceed.

Warning

This class will use an in-memory database for cache, unless a cache file is explicity defined.

Example#

>>> source = {'one': WindowParams(window=1)}
>>> params = WindowsParams(windows=source)
>>> windows = Windows(params, '-2s', 'now')
>>> [windows.ready('one') for x in range(4)]
[True, True, True, False]
param params:

Parameters used to instantiate the class.

param start:

Determine the start for scheduling window.

param stop:

Determine the ending for scheduling window.

param store:

Optional database path for keeping state.

param group:

Optional override for default group name.

property children: dict[str, Window]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

create(unique: str, params: WindowParams) → Window[source]#

Create a new window using the provided input parameters.

Parameters:

  • unique – Unique identifier for the related child.

  • params – Parameters used to instantiate the class.

Returns:

Newly constructed instance of related class.

delete(unique: str) → None[source]#

Delete the window from the internal dictionary reference.

Note

This is a graceful method, will not raise exception when the provided unique value does not exist.

Parameters:

unique – Unique identifier for the related child.

property group: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property params: WindowsParams#

Return the Pydantic model containing the configuration.

Returns:

Pydantic model containing the configuration.

pause(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

ready(unique: str, update: bool = True) → bool[source]#

Determine whether or not the appropriate time has passed.

Parameters:

  • unique – Unique identifier for the related child.

  • update – Determines whether or not time is updated.

Returns:

Boolean indicating whether enough time passed.

save_children() → None[source]#

Save the child caches from the attribute into database.

select(unique: str) → WindowsRecord | None[source]#

Return the record from within the table in the database.

Returns:

Record from within the table in the database.

property start: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property stop: Time#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store: str#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_engine: Engine#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_session: Session#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

property store_table: Type[WindowsTable]#

Return the value for the attribute from class instance.

Returns:

Value for the attribute from class instance.

update(unique: str, value: str | int | float | datetime | Time | None = None) → None[source]#

Update the window from the provided parasable time value.

Parameters:

  • unique – Unique identifier for the related child.

  • value – Override the time updated for window value.

pydantic model encommon.times.WindowsParams[source]#

Bases: BaseModel

Process and validate the core configuration parameters.

Show JSON schema
{
   "title": "WindowsParams",
   "description": "Process and validate the core configuration parameters.",
   "type": "object",
   "properties": {
      "windows": {
         "additionalProperties": {
            "$ref": "#/$defs/WindowParams"
         },
         "description": "Period for scheduling windows",
         "minProperties": 0,
         "title": "Windows",
         "type": "object"
      }
   },
   "$defs": {
      "WindowParams": {
         "additionalProperties": false,
         "description": "Process and validate the core configuration parameters.",
         "properties": {
            "window": {
               "anyOf": [
                  {
                     "type": "string"
                  },
                  {
                     "additionalProperties": {
                        "type": "integer"
                     },
                     "type": "object"
                  }
               ],
               "description": "Period for scheduling window",
               "minLength": 1,
               "title": "Window"
            },
            "start": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Determine when scope begins",
               "title": "Start"
            },
            "stop": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Determine when scope ends",
               "title": "Stop"
            },
            "anchor": {
               "anyOf": [
                  {
                     "minLength": 1,
                     "type": "string"
                  },
                  {
                     "type": "null"
                  }
               ],
               "default": null,
               "description": "Optional anchor of the window",
               "title": "Anchor"
            },
            "delay": {
               "description": "Time period of schedule delay",
               "minimum": 0,
               "title": "Delay",
               "type": "number"
            }
         },
         "required": [
            "window",
            "delay"
         ],
         "title": "WindowParams",
         "type": "object"
      }
   },
   "additionalProperties": false,
   "required": [
      "windows"
   ]
}

Config:

  • extra: str = forbid

Fields:
field windows: Annotated[dict[str, WindowParams], FieldInfo(annotation=NoneType, required=True, description='Period for scheduling windows', metadata=[MinLen(min_length=0)])] [Required]#

Period for scheduling windows

Constraints:

  • min_length = 0

encommon.times.findtz(tzname: str | None = None) → tzinfo[source]#

Return the located timezone object for the provided name.

Example#

>>> findtz('US/Central')
tzfile('/usr/share/zoneinfo/US/Central')
param tzname:

Name of the timezone associated to source.

returns:

Located timezone object for the provided name.

encommon.times.parse_time(source: str | int | float | datetime | Time | None = None, *, anchor: str | int | float | datetime | Time | None = None, format: str | None = None, tzname: str | None = None) → datetime[source]#

Parse provided time value with various supported formats.

Example

Description

‘now’

Uses the current time

None, ‘None’, ‘null’

Uses the current time

‘min’, float(‘-inf’)

Uses the minimum time

‘max’, float(‘inf’)

Uses the maximum time

object

Provide datetime or Time

2000-01-01T00:00:00Z

Uses strptime and dateutil

1645843626.2

Provide Unix epoch

‘-1h@h+5m’

Provide snaptime syntax

Example#

>>> parse_time('1/1/2000 12:00am')
datetime.datetime(2000, 1, 1, 0...

Example#

>>> parse_time(0)
datetime.datetime(1970, 1, 1, 0...
param source:

Time in various forms that will be parsed.

param anchor:

Optional relative time; for snap notation.

param format:

Optional format when source is timestamp.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

returns:

Python datetime object containing related time.

encommon.times.shift_time(notate: str, anchor: str | int | float | datetime | Time | None = None, *, tzname: str | None = None) → datetime[source]#

Shift provided time value using snaptime relative syntax.

Example#

>>> shift_time('-1h', anchor='1/1/2000 12:00am')
datetime.datetime(1999, 12, 31, 23...

Example#

>>> shift_time('-1h@d', anchor='1/1/2000 12:00am')
datetime.datetime(1999, 12, 31, 0...
param notate:

Syntax compatable using snaptime library.

param anchor:

Optional relative time; for snap notation.

param tzname:

Name of the timezone associated to anchor. This is not relevant in timezone included in anchor.

returns:

Python datetime object containing related time.

encommon.times.since_time(start: str | int | float | datetime | Time, stop: str | int | float | datetime | Time | None = None) → float[source]#

Determine the time in seconds occurring between values.

Example#

>>> start = parse_time(0)
>>> stop = parse_time(1)
>>> since_time(start, stop)
1.0
param start:

Time in various forms that will be parsed.

param stop:

Time in various forms that will be parsed.

returns:

Time in seconds occurring between the values.

encommon.times.string_time(source: str, *, formats: str | list[str] | None = None, tzname: str | None = None) → datetime[source]#

Parse provided time value with various supported formats.

Example#

>>> string_time('2000-01-01T00:00:00Z')
datetime.datetime(2000, 1, 1, 0...

Example#

>>> string_time('2000/01/01')
datetime.datetime(2000, 1, 1, 0...
param source:

Time in various forms that will be parsed.

param formats:

Various formats compatable with strptime.

param tzname:

Name of the timezone associated to source. This is not relevant in timezone included in source.

returns:

Python datetime object containing related time.

encommon.times.unitime(input: int | float | str) → int[source]#

Return the seconds in integer format for provided input.

Example#

>>> unitime('1d')
86400
>>> unitime('1y')
31536000
>>> unitime('1w3d4h')
878400
param input:

Input that will be converted into seconds.

returns:

Seconds in integer format for provided input.

On this page

This Page