Source code for snapm.manager._calendar
# Copyright Red Hat
#
# snapm/manager/calendar.py - Snapshot Manager CalendarSpec support
#
# This file is part of the snapm project.
#
# SPDX-License-Identifier: Apache-2.0
"""
Calendar event abstraction for Snapshot Manager
"""
from subprocess import run
from datetime import datetime, timedelta
import re
from snapm import SnapmCalloutError
_sd_analyze_calendar = ["systemd-analyze", "calendar"]
_ORIGINAL_FORM = "Original form"
_NORMALIZED_FORM = "Normalized form"
_NEXT_ELAPSE = "Next elapse"
_IN_UTC = "(in UTC)"
_FROM_NOW = "From now"
_NEVER = "never"
_TIME_FMT = "%a %Y-%m-%d %H:%M:%S %Z"
_TIME_REGEX = re.compile(r"\d{1,2}:\d{1,2}:\d{1,2}\.\d+")
USECS_PER_SEC = 1000000
[docs]class CalendarSpec:
"""
Class representing systemd CalendarSpec expressions.
"""
[docs] def _refresh(self):
"""
Refresh this ``CalendarSpec`` object's time-dependent properties.
"""
def strip_field(line):
"""
Strip the field name from the string ``line``.
"""
return line.split(": ", 1)[1].strip()
def parse_usecs(time):
"""
Parse the microseconds component of a time string and return it
as a timedelta object.
"""
parts = time.split()
for part in parts:
if _TIME_REGEX.match(part):
_, _, frac = part.partition(".")
frac, _, rep = frac.partition("/")
frac = int(frac.ljust(6, "0")) # Convert to microseconds
rep = float(rep) if rep else 0.0
return timedelta(
microseconds=USECS_PER_SEC * (rep + frac / USECS_PER_SEC)
)
return timedelta(0)
sd_cmd_args = _sd_analyze_calendar + [self._calendarspec]
sd_cmd = run(
sd_cmd_args,
encoding="utf8",
capture_output=True,
check=False,
)
if sd_cmd.returncode == 1:
raise ValueError(f"Invalid CalendarSpec expression: {self._calendarspec}")
if sd_cmd.returncode != 0:
raise SnapmCalloutError(
f"Error calling systemd-analyze: {sd_cmd.stderr.decode('utf8')}"
)
carry_usecs = timedelta(0)
for line in sd_cmd.stdout.splitlines():
line = line.strip()
if line.startswith(_ORIGINAL_FORM):
continue
if line.startswith(_NORMALIZED_FORM):
self.normalized = strip_field(line)
carry_usecs = parse_usecs(self.normalized)
if line.startswith(_NEXT_ELAPSE):
date_str = strip_field(line)
if date_str == _NEVER:
self._next_elapse = None
self._in_utc = None
self._from_now = _NEVER
continue
self._next_elapse = datetime.strptime(date_str, _TIME_FMT) + carry_usecs
if date_str.endswith("UTC"):
self._in_utc = True
if line.startswith(_IN_UTC):
date_str = strip_field(line)
self._in_utc = datetime.strptime(date_str, _TIME_FMT) + carry_usecs
if line.startswith(_FROM_NOW):
self._from_now = strip_field(line)
[docs] def _cond_refresh(self):
"""
Refresh this ``CalendarSpec`` object's time-dependent properties if the
current ``next_elapse`` value is in the past.
"""
# CalendarSpec instances that never occur do not need to be refreshed.
if self._next_elapse is not None and self._next_elapse < datetime.now():
self._refresh()
[docs] def __init__(self, calendarspec: str):
"""
Validate and parse a systemd calendarspec expression into a
CalendarSpec object.
:param calendarspec: A string containing an calendarspec expression.
:raises: ``ValueError`` if ``calendarspec`` is not a valid calendarspec
expression.
"""
self._calendarspec = calendarspec
self._refresh()
@property
def next_elapse(self):
"""
Return the next elapse time for this ``CalendarSpec`` object as an
instance of ``datetime.datetime``.
:returns: The next elapse time as a ``datetime`` object.
:rtype: ``datetime.datetime``
"""
self._cond_refresh()
return self._next_elapse
@property
def in_utc(self):
"""
Return the next elapse time for this ``CalendarSpec`` object as an
instance of ``datetime.datetime`` in UTC.
:returns: The next elapse time as a UTC ``datetime`` object.
:rtype: ``datetime.datetime``
"""
self._cond_refresh()
self._cond_refresh()
return self._in_utc
@property
def from_now(self):
"""
Return a string representation of the time remaining until this
``CalendarSpec`` next elapses.
:returns: The time remaining until the next elapse as a string.
:rtype: str
"""
self._cond_refresh()
return self._from_now
@property
def occurs(self):
"""
``True`` if this ``CalendarSpec`` object's calendar expression will
occur in the future, or ``False`` otherwise.
"""
return (
self.next_elapse is not None
and self.in_utc is not None
and self.from_now != _NEVER
)
@property
def original(self):
"""
The original form of this ``CalendarSpec`` as a string.
"""
return self._calendarspec
[docs] def __str__(self):
"""
Return a string representation of this ``CalendarSpec`` instance.
"""
return self._calendarspec
[docs] def __repr__(self):
"""
Return a string representation of this ``CalendarSpec`` instance in the
form of a call to the CalendarSpec initializer.
"""
return f'CalendarSpec("{str(self)}")'