def get_start_end_times()

in model/topology/heron/queueing_theory.py [0:0]

• 33 lines of code
• 7 McCabe index (conditional complexity)

def get_start_end_times(**kwargs) -> (dt.datetime, dt.datetime):
    if "start" in kwargs and "end" in kwargs:
        start_ts: int = int(kwargs["start"])
        start: dt.datetime = dt.datetime.utcfromtimestamp(start_ts)
        end_ts: int = int(kwargs["end"])
        end: dt.datetime = dt.datetime.utcfromtimestamp(end_ts)
        LOG.info("Start and end time stamps supplied, using metric "
                 "gathering period from %s to %s", start.isoformat(),
                 end.isoformat())
    elif "start" in kwargs and "end" not in kwargs:
        end = dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)
        start_ts = int(kwargs["start"])
        start = dt.datetime.utcfromtimestamp(start_ts)
        LOG.info("Only start time (%s) was supplied. Setting end time to "
                 "UTC now: %s", start.isoformat(), end.isoformat())
    elif "source_hours" in kwargs:
        end = dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)
        start = end - dt.timedelta(hours=int(kwargs["source_hours"]))
        LOG.info("Source hours provided, using metric gathering period "
                 "from %s to %s", start.isoformat(), end.isoformat())
    elif "source_mins" in kwargs:
        end = dt.datetime.utcnow().replace(tzinfo=dt.timezone.utc)
        start = end - dt.timedelta(minutes=int(kwargs["source_mins"]))
        LOG.info("Source mins provided, using metric gathering period "
                 "from %s to %s", start.isoformat(), end.isoformat())

    else:
        err_msg: str = ("Neither 'start', 'end' or 'source_hours' or 'source_mins' "
                        "key word arguments were supplied. Either 'start',"
                        " 'start' and 'end' or 'source_hours' should be "
                        "provided")
        LOG.error(err_msg)
        raise RuntimeError(err_msg)

    return start, end