Skip to content

dp3.common.utils

auxiliary/utility functions and classes

entity_expired 

entity_expired(utcnow: datetime, master_document: dict)

Check if entity is expired (all TTLs are in the past)

Source code in dp3/common/utils.py 
14
15
16
17
18
def entity_expired(utcnow: datetime, master_document: dict):
    """Check if entity is expired (all TTLs are in the past)"""
    if "#ttl" not in master_document:
        return False
    return all(ttl < utcnow for ttl in master_document["#ttl"].values())

parse_rfc_time 

parse_rfc_time(time_str)

Parse time in RFC 3339 format and return it as naive datetime in UTC.

Timezone specification is optional (UTC is assumed when none is specified).

Source code in dp3/common/utils.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def parse_rfc_time(time_str):
    """
    Parse time in RFC 3339 format and return it as naive datetime in UTC.

    Timezone specification is optional (UTC is assumed when none is specified).
    """
    res = timestamp_re.match(time_str)
    if res is not None:
        year, month, day, hour, minute, second = (int(n or 0) for n in res.group(*range(1, 7)))
        us_str = (res.group(7) or "0")[:6].ljust(6, "0")
        us = int(us_str)
        zonestr = res.group(8)
        zoneoffset = 0 if zonestr in (None, "z", "Z") else int(zonestr[:3]) * 60 + int(zonestr[4:6])
        zonediff = datetime.timedelta(minutes=zoneoffset)
        return datetime.datetime(year, month, day, hour, minute, second, us) - zonediff
    else:
        raise ValueError("Wrong timestamp format")

parse_time_duration 

parse_time_duration(duration_string: Union[str, int, datetime.timedelta]) -> datetime.timedelta

Parse duration in format (or just "0").

Return datetime.timedelta

Source code in dp3/common/utils.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def parse_time_duration(duration_string: Union[str, int, datetime.timedelta]) -> datetime.timedelta:
    """
    Parse duration in format <num><s/m/h/d> (or just "0").

    Return datetime.timedelta
    """
    # if it's already timedelta, just return it unchanged
    if isinstance(duration_string, datetime.timedelta):
        return duration_string
    # if number is passed, consider it number of seconds
    if isinstance(duration_string, (int, float)):
        return datetime.timedelta(seconds=duration_string)

    d = 0
    h = 0
    m = 0
    s = 0

    if duration_string == "0":
        pass
    elif duration_string[-1] == "d":
        d = int(duration_string[:-1])
    elif duration_string[-1] == "h":
        h = int(duration_string[:-1])
    elif duration_string[-1] == "m":
        m = int(duration_string[:-1])
    elif duration_string[-1] == "s":
        s = int(duration_string[:-1])
    else:
        raise ValueError("Invalid time duration string")

    return datetime.timedelta(days=d, hours=h, minutes=m, seconds=s)

conv_to_json 

conv_to_json(obj)

Convert special types to JSON (use as "default" param of json.dumps)

Supported types/objects: - datetime - timedelta

Source code in dp3/common/utils.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def conv_to_json(obj):
    """Convert special types to JSON (use as "default" param of json.dumps)

    Supported types/objects:
    - datetime
    - timedelta
    """
    if isinstance(obj, datetime.datetime):
        if obj.tzinfo:
            raise NotImplementedError(
                "Can't serialize timezone-aware datetime object "
                "(DP3 policy is to use naive datetimes in UTC everywhere)"
            )
        return {"$datetime": obj.strftime("%Y-%m-%dT%H:%M:%S.%f")}
    if isinstance(obj, datetime.timedelta):
        return {"$timedelta": f"{obj.days},{obj.seconds},{obj.microseconds}"}
    raise TypeError(f"{repr(obj)}%r is not JSON serializable")

conv_from_json 

conv_from_json(dct)

Convert special JSON keys created by conv_to_json back to Python objects (use as "object_hook" param of json.loads)

Supported types/objects: - datetime - timedelta

Source code in dp3/common/utils.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def conv_from_json(dct):
    """Convert special JSON keys created by conv_to_json back to Python objects
    (use as "object_hook" param of json.loads)

    Supported types/objects:
    - datetime
    - timedelta
    """
    if "$datetime" in dct:
        val = dct["$datetime"]
        return datetime.datetime.strptime(val, "%Y-%m-%dT%H:%M:%S.%f")
    if "$timedelta" in dct:
        days, seconds, microseconds = dct["$timedelta"].split(",")
        return datetime.timedelta(int(days), int(seconds), int(microseconds))
    return dct

get_func_name 

get_func_name(func_or_method)

Get name of function or method as pretty string.

Source code in dp3/common/utils.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def get_func_name(func_or_method):
    """Get name of function or method as pretty string."""
    if isinstance(func_or_method, partial):
        wrapper = "partial({name}, {args})"
        args = [str(a) for a in func_or_method.args]
        args.extend(f"{k}={v}" for k, v in func_or_method.keywords.items())
        args = ", ".join(args)
        func_or_method = func_or_method.func
    else:
        wrapper = "{name}{args}"
        args = ""

    try:
        fname = func_or_method.__func__.__qualname__
    except AttributeError:
        try:
            fname = func_or_method.__name__
        except AttributeError:
            fname = str(func_or_method)

    try:
        module = func_or_method.__module__
    except AttributeError:
        return fname
    return wrapper.format(name=f"{module}.{fname}", args=args)