Singularity Restart Count [Python]

restart_count.py

#!/usr/bin/env python

import re
import arrow
import argparse
import urllib.parse

import requests

CLUSTERS = {
    # <REDACTED>
}

RELATIVE_RE = re.compile(
    r"^now(?P<offset>(?P<dir>\+|-)(?P<amount>\d+)(?P<interval>[smhdMwy]))?$",
    re.I
)

INTERVALS = {
    "s": "seconds",
    "m": "minutes",
    "h": "hours",
    "d": "days",
    "w": "weeks",
    "M": "months",
    "y": "years",
}


def parse_relative(relative):
    m = RELATIVE_RE.match(relative)
    if not m:
        raise ValueError(f"invalid relative timestamp: {relative}")
    offset = m.groupdict()
    ts = arrow.utcnow()
    if offset["offset"]:
        interval = INTERVALS[offset["interval"]]
        amount = int(offset["amount"])
        if offset["dir"] == "-":
            amount *= -1
        ts = ts.shift(**{interval: amount})
    return ts

def compact_vals(d):
    return { k: v for k, v in d.items() if v is not None }

def task_history(root_url, request_id, since=None):
    url = urllib.parse.urljoin(root_url, f"/api/history/request/{request_id}/tasks")
    params = compact_vals({
        "startedAfter": since.timestamp * 1000 if since else None,
    })
    return requests.get(url, params=params).json()

def parse_args(args=None, parse=True):
    parser = argparse.ArgumentParser(description="Test")

    parser.add_argument("cluster", help="Singularity Cluster", choices=CLUSTERS.keys())
    parser.add_argument("request_id", help="Singularity Request")
    parser.add_argument("-s", "--since", type=parse_relative)

    res = parser.parse_args(args) if parse else None
    return parser, res

def main():
    _parser, args = parse_args()

    cluster = CLUSTERS[args.cluster]
    history = task_history(cluster, args.request_id, args.since)

    print(f"Number of task state changes for {args.request_id} in {args.cluster} since {args.since.humanize()}: {len(history)}")


if __name__ == "__main__":
    main()