¡@

Home 

OpenStack Study: timeutils.py

OpenStack Index

**** CubicPower OpenStack Study ****

def isotime(at=None, subsecond=False):

    """Stringify time in ISO 8601 format."""

    if not at:

        at = utcnow()

    st = at.strftime(_ISO8601_TIME_FORMAT

                     if not subsecond

                     else _ISO8601_TIME_FORMAT_SUBSECOND)

    tz = at.tzinfo.tzname(None) if at.tzinfo else 'UTC'

    st += ('Z' if tz == 'UTC' else tz)

    return st

**** CubicPower OpenStack Study ****

def parse_isotime(timestr):

    """Parse time from ISO 8601 format."""

    try:

        return iso8601.parse_date(timestr)

    except iso8601.ParseError as e:

        raise ValueError(six.text_type(e))

    except TypeError as e:

        raise ValueError(six.text_type(e))

**** CubicPower OpenStack Study ****

def strtime(at=None, fmt=PERFECT_TIME_FORMAT):

    """Returns formatted utcnow."""

    if not at:

        at = utcnow()

    return at.strftime(fmt)

**** CubicPower OpenStack Study ****

def parse_strtime(timestr, fmt=PERFECT_TIME_FORMAT):

    """Turn a formatted time back into a datetime."""

    return datetime.datetime.strptime(timestr, fmt)

**** CubicPower OpenStack Study ****

def normalize_time(timestamp):

    """Normalize time in arbitrary timezone to UTC naive object."""

    offset = timestamp.utcoffset()

    if offset is None:

        return timestamp

    return timestamp.replace(tzinfo=None) - offset

**** CubicPower OpenStack Study ****

def is_older_than(before, seconds):

    """Return True if before is older than seconds."""

    if isinstance(before, six.string_types):

        before = parse_strtime(before).replace(tzinfo=None)

    return utcnow() - before > datetime.timedelta(seconds=seconds)

**** CubicPower OpenStack Study ****

def is_newer_than(after, seconds):

    """Return True if after is newer than seconds."""

    if isinstance(after, six.string_types):

        after = parse_strtime(after).replace(tzinfo=None)

    return after - utcnow() > datetime.timedelta(seconds=seconds)

**** CubicPower OpenStack Study ****

def utcnow_ts():

    """Timestamp version of our utcnow function."""

    if utcnow.override_time is None:

        # NOTE(kgriffs): This is several times faster

        # than going through calendar.timegm(...)

        return int(time.time())

    return calendar.timegm(utcnow().timetuple())

**** CubicPower OpenStack Study ****

def utcnow():

    """Overridable version of utils.utcnow."""

    if utcnow.override_time:

        try:

            return utcnow.override_time.pop(0)

        except AttributeError:

            return utcnow.override_time

    return datetime.datetime.utcnow()

**** CubicPower OpenStack Study ****

def iso8601_from_timestamp(timestamp):

    """Returns a iso8601 formated date from timestamp."""

    return isotime(datetime.datetime.utcfromtimestamp(timestamp))

utcnow.override_time = None

**** CubicPower OpenStack Study ****

def set_time_override(override_time=None):

    """Overrides utils.utcnow.

    Make it return a constant time or a list thereof, one at a time.

    :param override_time: datetime instance or list thereof. If not

                          given, defaults to the current UTC time.

    """

    utcnow.override_time = override_time or datetime.datetime.utcnow()

**** CubicPower OpenStack Study ****

def advance_time_delta(timedelta):

    """Advance overridden time using a datetime.timedelta."""

    assert(not utcnow.override_time is None)

    try:

        for dt in utcnow.override_time:

            dt += timedelta

    except TypeError:

        utcnow.override_time += timedelta

**** CubicPower OpenStack Study ****

def advance_time_seconds(seconds):

    """Advance overridden time by seconds."""

    advance_time_delta(datetime.timedelta(0, seconds))

**** CubicPower OpenStack Study ****

def clear_time_override():

    """Remove the overridden time."""

    utcnow.override_time = None

**** CubicPower OpenStack Study ****

def marshall_now(now=None):

    """Make an rpc-safe datetime with microseconds.

    Note: tzinfo is stripped, but not required for relative times.

    """

    if not now:

        now = utcnow()

    return dict(day=now.day, month=now.month, year=now.year, hour=now.hour,

                minute=now.minute, second=now.second,

                microsecond=now.microsecond)

**** CubicPower OpenStack Study ****

def unmarshall_time(tyme):

    """Unmarshall a datetime dict."""

    return datetime.datetime(day=tyme['day'],

                             month=tyme['month'],

                             year=tyme['year'],

                             hour=tyme['hour'],

                             minute=tyme['minute'],

                             second=tyme['second'],

                             microsecond=tyme['microsecond'])

**** CubicPower OpenStack Study ****

def delta_seconds(before, after):

    """Return the difference between two timing objects.

    Compute the difference in seconds between two date, time, or

    datetime objects (as a float, to microsecond resolution).

    """

    delta = after - before

    return total_seconds(delta)

**** CubicPower OpenStack Study ****

def total_seconds(delta):

    """Return the total seconds of datetime.timedelta object.

    Compute total seconds of datetime.timedelta, datetime.timedelta

    doesn't have method total_seconds in Python2.6, calculate it manually.

    """

    try:

        return delta.total_seconds()

    except AttributeError:

        return ((delta.days * 24 * 3600) + delta.seconds +

                float(delta.microseconds) / (10 ** 6))

**** CubicPower OpenStack Study ****

def is_soon(dt, window):

    """Determines if time is going to happen in the next window seconds.

    :params dt: the time

    :params window: minimum seconds to remain to consider the time not soon

    :return: True if expiration is within the given duration

    """

    soon = (utcnow() + datetime.timedelta(seconds=window))

    return normalize_time(dt) <= soon