utils module¶
This module implements miscellaneous utility functions and classes.
- class swpt_pythonlib.utils.Int64Converter(map: Map, *args: t.Any, **kwargs: t.Any)¶
Flask URL converter for signed 64-bit integers.
The converter can be registered with the Flask app like this:
from flask import Flask from swpt_pythonlib.utils import Int64Converter app = Flask(__name__) app.url_map.converters['i64'] = Int64Converter
- class swpt_pythonlib.utils.Seqnum(value: int)¶
A signed 32-bit integer seqnum value.
Comparisions beteen Seqnum instances correctly deal with the possible 32-bit integer wrapping.
- class swpt_pythonlib.utils.ShardingRealm(routing_key: str)¶
Holds a sharding key and a sharding mask.
This class allows to easily check if a given shard is responsible for a given creditor/debtor ID (or a debtor ID, creditor ID pair).
- Variables
realm_mask – Bitmask for the realm (a 32-bit unsigned integer).
realm – Bitprefix for the realm (a 32-bit unsigned integer)
parent_realm_mask – Bitmask for the parent realm.
parent_realm – Bitprefix for the parent realm.
- match(first: int, *rest: int, match_parent=False) bool¶
Return whether the shard is responsible for the passed sharding key.
Also, it is possible to check whether the parent shard would be responsible for the passed sharding key.
Example:
>>> r = ShardingRealm('1.#') >>> r.match(1) True >>> r.match(3) False >>> r.match(3, match_parent=True) # The parent shard is "#" True >>> r.match(1, 2) # (Debtor ID, Creditor ID) pair True
- match_str(s: str, match_parent=False) bool¶
Return whether the shard is responsible for the passed string-key.
Also, it is possible to check whether the parent shard would be responsible for the passed string-key.
Example:
>>> r = ShardingRealm('1.#') >>> r.match("65") True >>> r.match("59") False >>> r.match("59", match_parent=True) # The parent shard is "#" True
- swpt_pythonlib.utils.calc_bin_routing_key(first: int, *rest: int) str¶
Calculate a binary RabbitMQ routing key from one or more i64 numbers.
The binary routing key is calculated by taking the highest 24 bits, separated with dots, of the MD5 digest of the passed numbers. For example:
>>> calc_bin_routing_key(123) '1.1.1.1.1.1.0.0.0.0.0.1.0.0.0.0.0.1.1.0.0.0.1.1' >>> calc_bin_routing_key(-123) '1.1.0.0.0.0.1.1.1.1.1.1.1.1.1.0.1.0.1.0.1.1.1.1' >>> calc_bin_routing_key(123, 456) '0.0.0.0.1.0.0.0.0.1.0.0.0.1.0.0.0.0.1.1.0.1.0.0'
- swpt_pythonlib.utils.calc_iri_routing_key(iri: str) str¶
Calculate a binary RabbitMQ routing key from an IRI string.
The binary routing key is calculated by taking the highest 24 bits, separated with dots, of the MD5 digest of the UTF-8 serialization of the passed IRI string. For example:
>>> calc_iri_routing_key("https://example.com/iri") '0.1.0.0.0.0.1.1.1.1.0.1.0.1.0.0.1.1.0.0.0.1.0.1'
- swpt_pythonlib.utils.date_to_int24(d: date) int¶
Return a non-negative 24-bit integer derived from a date.
The passed date must not be before January 1st, 1970. The returned integer equals the number of days passed since January 1st, 1970.
- swpt_pythonlib.utils.get_config_value(key: str) Optional[str]¶
Get the value for the configuration variable with a name key.
The returned value is either a string or None. If there is a Flask application context, the app’s config will be checked first. If that fails, the environment will be checked next. If that fails too, None will be returned.
- swpt_pythonlib.utils.i64_to_hex_routing_key(n: int)¶
Calculate a hexadecimal RabbitMQ routing key from a i64 number.
The hexadecimal routing key is calculated by placing the 8 bytes of the number together, separated with dots. For example:
>>> i64_to_hex_routing_key(2) '00.00.00.00.00.00.00.02' >>> i64_to_hex_routing_key(-2) 'ff.ff.ff.ff.ff.ff.ff.fe'
- swpt_pythonlib.utils.i64_to_u64(value: int) int¶
Convert a signed 64-bit integer to unsigned 64-bit integer.
Raises ValueError if the value is not in the range of signed 64-bit integers.
- swpt_pythonlib.utils.is_later_event(event: Tuple[datetime, int], other_event: Tuple[Optional[datetime], Optional[int]]) bool¶
Return whether event is later than other_event.
Each of the passed events must be a (datetime, int) tuple. The datetime must be the event timestamp, and the int must be the event sequential number (32-bit signed integer, with eventual wrapping).
An event with a noticeably later timestamp (>= 1s) is always considered later than an event with an earlier timestamp. Only when the two timestamps are very close (< 1s), the sequential numbers of the events are compared. When the timestamp of other_event is None, event is considered as a later event.
Note that sequential numbers are compared with possible 32-bit signed integer wrapping in mind. For example, compared to 2147483647, -21474836478 is considered a later sequential number.