python_utils package¶

Submodules¶

python_utils.aio module¶

Asyncio equivalents of common synchronous helpers.

These bring itertools-style ergonomics to async for: acount is an async counter, while acontainer and adict collect an async iterable into a concrete container.

async python_utils.aio.acontainer(iterable: AsyncIterable[_T] | Callable[[...], AsyncIterable[_T]], container: type[tuple[_T, ...]]) → tuple[_T, ...][source]¶
async python_utils.aio.acontainer(iterable: AsyncIterable[_T] | Callable[[...], AsyncIterable[_T]], container: type[list[_T]] = list) → list[_T]
async python_utils.aio.acontainer(iterable: AsyncIterable[_T] | Callable[[...], AsyncIterable[_T]], container: type[set[_T]]) → set[_T]

Asyncio version of list()/set()/tuple()/etc() using an async for loop.

So instead of doing [item async for item in iterable] you can do await acontainer(iterable).

Parameters:
Return type:

Collection[_T]

async python_utils.aio.acount(start=0, step=1, delay=0, stop=None)[source]¶

Async equivalent of itertools.count().

Counts from start in steps of step, sleeping delay seconds between values, and stops once stop (when given) is reached.

Parameters:

  • start (_N) – First value to yield.

  • step (_N) – Amount added between successive values.

  • delay (float) – Seconds to asyncio.sleep between yields.

  • stop (_N | None) – Exclusive upper bound; None counts forever.

Yields:

The successive counter values.

Return type:

AsyncIterator[_N]

>>> async def demo():
...     return [i async for i in acount(stop=3)]
>>> asyncio.run(demo())
[0, 1, 2]
async python_utils.aio.adict(iterable, container=<class 'dict'>)[source]¶

Asyncio version of dict() using an async for loop.

So instead of doing {key: value async for key, value in iterable} you can do await adict(iterable).

Parameters:
Return type:

Mapping[_K, _V]

python_utils.containers module¶

This module provides custom container classes with enhanced functionality.

Classes:

CastedDictBase: Abstract base class for dictionaries that cast keys and
    values.
CastedDict: Dictionary that casts keys and values to specified types.
LazyCastedDict: Dictionary that lazily casts values to specified types upon
    access.
UniqueList: List that only allows unique values, with configurable behavior
    on duplicates.
SliceableDeque: Deque that supports slicing and enhanced equality checks.

Type Aliases:

KT: Type variable for dictionary keys.
VT: Type variable for dictionary values.
DT: Type alias for a dictionary with keys of type KT and values of type VT.
KT_cast: Type alias for a callable that casts dictionary keys.
VT_cast: Type alias for a callable that casts dictionary values.
HT: Type variable for hashable values in UniqueList.
T: Type variable for generic types.
DictUpdateArgs: Union type for arguments that can be used to update a
    dictionary.
OnDuplicate: Literal type for handling duplicate values in UniqueList.

Usage:

- CastedDict and LazyCastedDict can be used to create dictionaries with
    automatic type casting.
- UniqueList ensures all elements are unique and can raise an error on
    duplicates.
- SliceableDeque extends deque with slicing support and enhanced equality
    checks.

Examples

>>> d = CastedDict(int, int)
>>> d[1] = 2
>>> d['3'] = '4'
>>> d.update({'5': '6'})
>>> d.update([('7', '8')])
>>> d
{1: 2, 3: 4, 5: 6, 7: 8}

>>> l = UniqueList(1, 2, 3)
>>> l.append(4)
>>> l.append(4)
>>> l.insert(0, 4)
>>> l.insert(0, 5)
>>> l[1] = 10
>>> l
[5, 10, 2, 3, 4]

>>> d = SliceableDeque([1, 2, 3, 4, 5])
>>> d[1:4]
SliceableDeque([2, 3, 4])
class python_utils.containers.CastedDict(key_cast=None, value_cast=None, *args, **kwargs)[source]¶

Bases: CastedDictBase[KT, VT]

Custom dictionary that casts keys and values to the specified types.

Note that you can specify the types for mypy and type hinting with: CastedDict[int, int](int, int)

>>> d: CastedDict[int, int] = CastedDict(int, int)
>>> d[1] = 2
>>> d['3'] = '4'
>>> d.update({'5': '6'})
>>> d.update([('7', '8')])
>>> d
{1: 2, 3: 4, 5: 6, 7: 8}
>>> list(d.keys())
[1, 3, 5, 7]
>>> list(d)
[1, 3, 5, 7]
>>> list(d.values())
[2, 4, 6, 8]
>>> list(d.items())
[(1, 2), (3, 4), (5, 6), (7, 8)]
>>> d[3]
4

# Casts are optional and can be disabled by passing None as the cast >>> d = CastedDict() >>> d[1] = 2 >>> d[‘3’] = ‘4’ >>> d.update({‘5’: ‘6’}) >>> d.update([(‘7’, ‘8’)]) >>> d {1: 2, ‘3’: ‘4’, ‘5’: ‘6’, ‘7’: ‘8’}

Parameters:
class python_utils.containers.CastedDictBase(key_cast=None, value_cast=None, *args, **kwargs)[source]¶

Bases: dict[KT, VT], ABC

Abstract base class for dictionaries that cast keys and values.

Parameters:
_key_cast¶

Callable to cast dictionary keys.

Type:

KT_cast[KT]

_value_cast¶

Callable to cast dictionary values.

Type:

VT_cast[VT]

Methods:

__init__(key_cast: KT_cast[KT] = None, value_cast: VT_cast[VT] = None,
    *args: DictUpdateArgs[KT, VT], **kwargs: VT) -> None:
    Initializes the dictionary with optional key and value casting
    callables.
update(*args: DictUpdateArgs[typing.Any, typing.Any],
    **kwargs: typing.Any) -> None:
    Updates the dictionary with the given arguments.
__setitem__(key: typing.Any, value: typing.Any) -> None:
    Sets the item in the dictionary, casting the key if a key cast
    callable is provided.
update(*args, **kwargs)[source]¶

Updates the dictionary with the given arguments.

Parameters:

  • *args (DictUpdateArgs[Any, Any]) – Arguments to update the dictionary.

  • **kwargs (Any) – Keyword arguments to update the dictionary.

Return type:

None

python_utils.containers.DT¶

A type alias for a dictionary with keys of type KT and values of type VT.

alias of dict[KT, VT]

class python_utils.containers.HT¶

A type alias for the hashable values of the UniqueList

alias of TypeVar(‘HT’, bound=Hashable)

class python_utils.containers.KT¶

A type alias for a type that can be used as a key in a dictionary.

alias of TypeVar(‘KT’)

python_utils.containers.KT_cast = collections.abc.Callable[..., ~KT] | None¶

A type alias for the casted type of a dictionary key.

class python_utils.containers.LazyCastedDict(key_cast=None, value_cast=None, *args, **kwargs)[source]¶

Bases: CastedDictBase[KT, VT]

Custom dictionary that casts keys and lazily casts values to the specified types. Note that the values are cast only when they are accessed and are not cached between executions.

Note that you can specify the types for mypy and type hinting with: LazyCastedDict[int, int](int, int)

>>> d: LazyCastedDict[int, int] = LazyCastedDict(int, int)
>>> d[1] = 2
>>> d['3'] = '4'
>>> d.update({'5': '6'})
>>> d.update([('7', '8')])
>>> d
{1: 2, 3: '4', 5: '6', 7: '8'}
>>> list(d.keys())
[1, 3, 5, 7]
>>> list(d)
[1, 3, 5, 7]
>>> list(d.values())
[2, 4, 6, 8]
>>> list(d.items())
[(1, 2), (3, 4), (5, 6), (7, 8)]
>>> d[3]
4

# Casts are optional and can be disabled by passing None as the cast >>> d = LazyCastedDict() >>> d[1] = 2 >>> d[‘3’] = ‘4’ >>> d.update({‘5’: ‘6’}) >>> d.update([(‘7’, ‘8’)]) >>> d {1: 2, ‘3’: ‘4’, ‘5’: ‘6’, ‘7’: ‘8’} >>> list(d.keys()) [1, ‘3’, ‘5’, ‘7’] >>> list(d.values()) [2, ‘4’, ‘6’, ‘8’]

>>> list(d.items())
[(1, 2), ('3', '4'), ('5', '6'), ('7', '8')]
>>> d['3']
'4'
Parameters:
items()[source]¶

Returns a generator of the dictionary’s items, casting the values if a value cast callable is provided.

Yields:

Generator[tuple[KT, VT], None, None] –

A generator of

the dictionary’s items.

Return type:

Generator[tuple[KT, VT], None, None]

values()[source]¶

Returns a generator of the dictionary’s values, casting the values if a value cast callable is provided.

Yields:

Generator[VT, None, None] –

A generator of the dictionary’s

values.

Return type:

Generator[VT, None, None]

python_utils.containers.OnDuplicate¶

silently 'ignore' or 'raise'.

Type:

Policy for UniqueList duplicates

alias of Literal[‘ignore’, ‘raise’]

class python_utils.containers.SliceableDeque[source]¶

Bases: Generic[T], deque[T]

A deque that supports slicing and enhanced equality checks.

Methods:

__getitem__(index: typing.SupportsIndex | slice) ->
    T | 'SliceableDeque[T]':
    Returns the item or slice at the given index.
__eq__(other: typing.Any) -> bool:
    Checks equality with another object, allowing for comparison with
     lists, tuples, and sets.
pop(index: int = -1) -> T:
    Removes and returns the item at the given index. Only supports
    index 0 and the last index.
pop(index=-1)[source]¶

Removes and returns the item at the given index. Only supports index 0 and the last index.

Parameters:

  • index (int, optional) – The index of the item to remove. Defaults to

  • -1.

Returns:

The removed item.

Return type:

T

Raises:

IndexError – If the index is not 0 or the last index.

Examples

>>> d = SliceableDeque([1, 2, 3])
>>> d.pop(0)
1
>>> d.pop()
3
class python_utils.containers.T¶

A type alias for a regular generic type

alias of TypeVar(‘T’)

class python_utils.containers.UniqueList(*args, on_duplicate='ignore')[source]¶

Bases: list[HT]

A list that only allows unique values. Duplicate values are ignored by default, but can be configured to raise an exception instead.

>>> l = UniqueList(1, 2, 3)
>>> l.append(4)
>>> l.append(4)
>>> l.insert(0, 4)
>>> l.insert(0, 5)
>>> l[1] = 10
>>> l
[5, 10, 2, 3, 4]

>>> l = UniqueList(1, 2, 3, on_duplicate='raise')
>>> l.append(4)
>>> l.append(4)
Traceback (most recent call last):
...
ValueError: Duplicate value: 4
>>> l.insert(0, 4)
Traceback (most recent call last):
...
ValueError: Duplicate value: 4
>>> 4 in l
True
>>> l[0]
1
>>> l[1] = 4
Traceback (most recent call last):
...
ValueError: Duplicate value: 4
Parameters:

  • args (HT)

  • on_duplicate (Literal['ignore', 'raise'])

append(value)[source]¶

Appends a value to the list, ensuring uniqueness.

Parameters:

value (HT) – The value to append.

Raises:

ValueError – If the value is a duplicate and on_duplicate is set to ‘raise’.

Return type:

None

insert(index, value)[source]¶

Inserts a value at the specified index, ensuring uniqueness.

Parameters:

  • index (SupportsIndex) – The index to insert the value at.

  • value (HT) – The value to insert.

Raises:

ValueError – If the value is a duplicate and on_duplicate is set to ‘raise’.

Return type:

None

class python_utils.containers.VT¶

A type alias for a type that can be used as a value in a dictionary.

alias of TypeVar(‘VT’)

python_utils.containers.VT_cast = collections.abc.Callable[..., ~VT] | None¶

A type alias for the casted type of a dictionary value.

python_utils.converters module¶

This module provides utility functions for type conversion.

Functions:

to_int: Convert a string to an integer with optional regexp matching.
to_float: Convert a string to a float with optional regexp matching.
to_unicode: Convert objects to Unicode strings.
to_str: Convert objects to byte strings.
scale_1024: Scale a number down to a suitable size (powers of 1024).
remap: Remap a value from one range to another.
python_utils.converters.remap(value: Decimal, old_min: Decimal | float, old_max: Decimal | float, new_min: Decimal | float, new_max: Decimal | float) → Decimal[source]¶
python_utils.converters.remap(value: Decimal | float, old_min: Decimal, old_max: Decimal | float, new_min: Decimal | float, new_max: Decimal | float) → Decimal
python_utils.converters.remap(value: Decimal | float, old_min: Decimal | float, old_max: Decimal, new_min: Decimal | float, new_max: Decimal | float) → Decimal
python_utils.converters.remap(value: Decimal | float, old_min: Decimal | float, old_max: Decimal | float, new_min: Decimal, new_max: Decimal | float) → Decimal
python_utils.converters.remap(value: Decimal | float, old_min: Decimal | float, old_max: Decimal | float, new_min: Decimal | float, new_max: Decimal) → Decimal
python_utils.converters.remap(value: float, old_min: float, old_max: float, new_min: float, new_max: float) → float

Remap a value from one range into another.

>>> remap(500, 0, 1000, 0, 100)
50
>>> remap(250.0, 0.0, 1000.0, 0.0, 100.0)
25.0
>>> remap(-75, -100, 0, -1000, 0)
-750
>>> remap(33, 0, 100, -500, 500)
-170
>>> remap(decimal.Decimal('250.0'), 0.0, 1000.0, 0.0, 100.0)
Decimal('25.0')

This is a great use case example. Take an AVR that has dB values the minimum being -80dB and the maximum being 10dB and you want to convert volume percent to the equivalent in that dB range

>>> remap(46.0, 0.0, 100.0, -80.0, 10.0)
-38.6

I added using decimal.Decimal so floating point math errors can be avoided. Here is an example of a floating point math error >>> 0.1 + 0.1 + 0.1 0.30000000000000004

If floating point remaps need to be done my suggestion is to pass at least one parameter as a decimal.Decimal. This will ensure that the output from this function is accurate. I left passing floats for backwards compatibility and there is no conversion done from float to decimal.Decimal unless one of the passed parameters has a type of decimal.Decimal. This will ensure that any existing code that uses this function will work exactly how it has in the past.

Some edge cases to test >>> remap(1, 0, 0, 1, 2) Traceback (most recent call last): … ValueError: Input range (0-0) is empty

>>> remap(1, 1, 2, 0, 0)
Traceback (most recent call last):
...
ValueError: Output range (0-0) is empty
Parameters:
Return type:

_TN

Returns: int, float, decimal.Decimal: Value that has been re-ranged. If

any of the parameters passed is a decimal.Decimal, all of the parameters will be converted to decimal.Decimal. The same thing also happens if one of the parameters is a float. Otherwise, all parameters will get converted into an int. Technically, you can pass a str of an integer and it will get converted. The returned value type will be decimal.Decimal if any of the passed parameters are decimal.Decimal, the return type will be float if any of the passed parameters are a float, otherwise the returned type will be int.

python_utils.converters.scale_1024(x, n_prefixes)[source]¶

Scale a number down to a suitable size, based on powers of 1024.

Returns the scaled number and the power of 1024 used.

Use to format numbers of bytes to KiB, MiB, etc.

>>> scale_1024(310, 3)
(310.0, 0)
>>> scale_1024(2048, 3)
(2.0, 1)
>>> scale_1024(0, 2)
(0.0, 0)
>>> scale_1024(0.5, 2)
(0.5, 0)
>>> scale_1024(1, 2)
(1.0, 0)
Parameters:
Return type:

tuple[int | float, int | float]

python_utils.converters.to_float(input_, default=0, exception=(<class 'ValueError'>, <class 'TypeError'>), regexp=None)[source]¶

Convert the given input_ to an integer or return default.

When trying to convert the exceptions given in the exception parameter are automatically caught and the default will be returned.

The regexp parameter allows for a regular expression to find the digits in a string. When True it will automatically match any digit in the string. When a (regexp) object (has a search method) is given, that will be used. When a string is given, re.compile will be run over it first

The last group of the regexp will be used as value

>>> '%.2f' % to_float('abc')
'0.00'
>>> '%.2f' % to_float('1')
'1.00'
>>> '%.2f' % to_float('abc123.456', regexp=True)
'123.46'
>>> '%.2f' % to_float('abc123', regexp=True)
'123.00'
>>> '%.2f' % to_float('abc0.456', regexp=True)
'0.46'
>>> '%.2f' % to_float('abc123.456', regexp=re.compile(r'(\d+\.\d+)'))
'123.46'
>>> '%.2f' % to_float('123.456abc', regexp=re.compile(r'(\d+\.\d+)'))
'123.46'
>>> '%.2f' % to_float('abc123.46abc', regexp=re.compile(r'(\d+\.\d+)'))
'123.46'
>>> '%.2f' % to_float('abc123abc456', regexp=re.compile(r'(\d+(\.\d+|))'))
'123.00'
>>> '%.2f' % to_float('abc', regexp=r'(\d+)')
'0.00'
>>> '%.2f' % to_float('abc123', regexp=r'(\d+)')
'123.00'
>>> '%.2f' % to_float('123abc', regexp=r'(\d+)')
'123.00'
>>> '%.2f' % to_float('abc123abc', regexp=r'(\d+)')
'123.00'
>>> '%.2f' % to_float('abc123abc456', regexp=r'(\d+)')
'123.00'
>>> '%.2f' % to_float('1234', default=1)
'1234.00'
>>> '%.2f' % to_float('abc', default=1)
'1.00'
>>> '%.2f' % to_float('abc', regexp=123)
Traceback (most recent call last):
...
TypeError: unknown argument for regexp parameter
Parameters:
Return type:

int | float

python_utils.converters.to_int(input_=None, default=0, exception=(<class 'ValueError'>, <class 'TypeError'>), regexp=None)[source]¶

Convert the given input to an integer or return default.

When trying to convert the exceptions given in the exception parameter are automatically caught and the default will be returned.

The regexp parameter allows for a regular expression to find the digits in a string. When True it will automatically match any digit in the string. When a (regexp) object (has a search method) is given, that will be used. WHen a string is given, re.compile will be run over it first

The last group of the regexp will be used as value

>>> to_int('abc')
0
>>> to_int('1')
1
>>> to_int('')
0
>>> to_int()
0
>>> to_int('abc123')
0
>>> to_int('123abc')
0
>>> to_int('abc123', regexp=True)
123
>>> to_int('123abc', regexp=True)
123
>>> to_int('abc123abc', regexp=True)
123
>>> to_int('abc123abc456', regexp=True)
123
>>> to_int('abc123', regexp=re.compile(r'(\d+)'))
123
>>> to_int('123abc', regexp=re.compile(r'(\d+)'))
123
>>> to_int('abc123abc', regexp=re.compile(r'(\d+)'))
123
>>> to_int('abc123abc456', regexp=re.compile(r'(\d+)'))
123
>>> to_int('abc123', regexp=r'(\d+)')
123
>>> to_int('123abc', regexp=r'(\d+)')
123
>>> to_int('abc', regexp=r'(\d+)')
0
>>> to_int('abc123abc', regexp=r'(\d+)')
123
>>> to_int('abc123abc456', regexp=r'(\d+)')
123
>>> to_int('1234', default=1)
1234
>>> to_int('abc', default=1)
1
>>> to_int('abc', regexp=123)
Traceback (most recent call last):
...
TypeError: unknown argument for regexp parameter: 123
Parameters:
Return type:

int

python_utils.converters.to_str(input_, encoding='utf-8', errors='replace')[source]¶

Convert objects to string, encodes to the given encoding.

Return type:

str

Parameters:
>>> to_str('a')
b'a'
>>> to_str('a')
b'a'
>>> to_str(b'a')
b'a'
>>> class Foo(object):
...     __str__ = lambda s: 'a'
>>> to_str(Foo())
'a'
>>> to_str(Foo)
"<class 'python_utils.converters.Foo'>"
python_utils.converters.to_unicode(input_, encoding='utf-8', errors='replace')[source]¶

Convert objects to unicode, if needed decodes string with the given encoding and errors settings.

Return type:

str

Parameters:
>>> to_unicode(b'a')
'a'
>>> to_unicode('a')
'a'
>>> to_unicode('a')
'a'
>>> class Foo(object):
...     __str__ = lambda s: 'a'
>>> to_unicode(Foo())
'a'
>>> to_unicode(Foo)
"<class 'python_utils.converters.Foo'>"

python_utils.decorators module¶

This module provides various utility decorators for Python functions and methods.

The decorators include:

  1. set_attributes: Sets attributes on functions and classes.

  2. listify: Converts any generator to a list or other collection.

  3. sample: Limits calls to a function based on a sample rate.

  4. wraps_classmethod: Wraps classmethods with type info from a regular method.

Each decorator is designed to enhance the functionality of Python functions and methods in a simple and reusable manner.

python_utils.decorators.listify(collection=<class 'list'>, allow_empty=True)[source]¶

Convert any generator to a list or other type of collection.

>>> @listify()
... def generator():
...     yield 1
...     yield 2
...     yield 3

>>> generator()
[1, 2, 3]

>>> @listify()
... def empty_generator():
...     pass

>>> empty_generator()
[]

>>> @listify(allow_empty=False)
... def empty_generator_not_allowed():
...     pass

>>> empty_generator_not_allowed()
Traceback (most recent call last):
...
TypeError: ... `allow_empty` is `False`

>>> @listify(collection=set)
... def set_generator():
...     yield 1
...     yield 1
...     yield 2

>>> set_generator()
{1, 2}

>>> @listify(collection=dict)
... def dict_generator():
...     yield 'a', 1
...     yield 'b', 2

>>> dict_generator()
{'a': 1, 'b': 2}
Parameters:
Return type:

Callable[[Callable[[…], Iterable[_T] | None]], Callable[[…], Collection[_T]]]

python_utils.decorators.sample(sample_rate)[source]¶

Limit calls to a function based on given sample rate. Number of calls to the function will be roughly equal to sample_rate percentage.

Usage:

>>> @sample(0.5)
... def demo_function(*args, **kwargs):
...     return 1

Calls to demo_function will be limited to 50% approximately.

Parameters:

sample_rate (float)

Return type:

Callable[[Callable[[~_P], _T]], Callable[[~_P], _T | None]]

python_utils.decorators.set_attributes(**kwargs)[source]¶

Decorator to set attributes on functions and classes.

A common usage for this pattern is the Django Admin where functions can get an optional short_description. To illustrate:

Example from the Django admin using this decorator: https://docs.djangoproject.com/en/3.0/ref/contrib/admin/#django.contrib.admin.ModelAdmin.list_display

Our simplified version:

>>> @set_attributes(short_description='Name')
... def upper_case_name(self, obj):
...     return ('%s %s' % (obj.first_name, obj.last_name)).upper()

The standard Django version:

>>> def upper_case_name(obj):
...     return ('%s %s' % (obj.first_name, obj.last_name)).upper()

>>> upper_case_name.short_description = 'Name'
Parameters:

kwargs (Any)

Return type:

Callable[[…], Any]

python_utils.decorators.wraps_classmethod(wrapped)[source]¶

Like functools.wraps, but for wrapping classmethods.

Copies the wrapped method’s metadata (name, docstring and annotations) onto the wrapper, so a classmethod wrapper carries the type information of the regular method it stands in for.

Parameters:

wrapped (Callable[[Concatenate[Any, ~_P]], _T]) – The method whose metadata should be copied onto the wrapper.

Returns:

A decorator that updates its wrapper with wrapped’s metadata.

Return type:

Callable[[Callable[[Concatenate[Any, ~_P]], _T]], Callable[[Concatenate[Any, ~_P]], _T]]

python_utils.exceptions module¶

This module provides utility functions for raising and reraising exceptions.

Functions:

raise_exception(exception_class, *args, **kwargs):
    Returns a function that raises an exception of the given type with
    the given arguments.

reraise(*args, **kwargs):
    Reraises the current exception.
python_utils.exceptions.raise_exception(exception_class, *args, **kwargs)[source]¶

Returns a function that raises an exception of the given type with the given arguments.

>>> raise_exception(ValueError, 'spam')('eggs')
Traceback (most recent call last):
    ...
ValueError: spam
Parameters:
Return type:

Callable[[…], None]

python_utils.exceptions.reraise(*args, **kwargs)[source]¶

Reraises the current exception.

This function seems useless, but it can be useful when you need to pass a callable to another function that raises an exception.

Parameters:
Return type:

Any

python_utils.formatters module¶

This module provides utility functions for formatting strings and dates.

Functions:

camel_to_underscore: Convert camel case naming to underscore/snake case.
apply_recursive: Apply a function to all keys in a scope recursively.
timesince: Return a string representing 'time since', e.g. 3 days ago.
python_utils.formatters.apply_recursive(function, data=None, **kwargs)[source]¶

Apply a function to all keys in a scope recursively.

>>> apply_recursive(camel_to_underscore, {'SpamEggsAndBacon': 'spam'})
{'spam_eggs_and_bacon': 'spam'}
>>> apply_recursive(
...     camel_to_underscore,
...     {
...         'SpamEggsAndBacon': {
...             'SpamEggsAndBacon': 'spam',
...         }
...     },
... )
{'spam_eggs_and_bacon': {'spam_eggs_and_bacon': 'spam'}}

>>> a = {'a_b_c': 123, 'def': {'DeF': 456}}
>>> b = apply_recursive(camel_to_underscore, a)
>>> b
{'a_b_c': 123, 'def': {'de_f': 456}}

>>> apply_recursive(camel_to_underscore, None)
Parameters:
Return type:

dict[str, Any] | None

python_utils.formatters.camel_to_underscore(name)[source]¶

Convert camel case style naming to underscore/snake case style naming.

If there are existing underscores they will be collapsed with the to-be-added underscores. Multiple consecutive capital letters will not be split except for the last one.

>>> camel_to_underscore('SpamEggsAndBacon')
'spam_eggs_and_bacon'
>>> camel_to_underscore('Spam_and_bacon')
'spam_and_bacon'
>>> camel_to_underscore('Spam_And_Bacon')
'spam_and_bacon'
>>> camel_to_underscore('__SpamAndBacon__')
'__spam_and_bacon__'
>>> camel_to_underscore('__SpamANDBacon__')
'__spam_and_bacon__'
Parameters:

name (str)

Return type:

str

python_utils.formatters.timesince(dt, default='just now')[source]¶

Returns string representing ‘time since’ e.g. 3 days ago, 5 hours ago etc.

>>> now = datetime.datetime.now()
>>> timesince(now)
'just now'
>>> timesince(now - datetime.timedelta(seconds=1))
'1 second ago'
>>> timesince(now - datetime.timedelta(seconds=2))
'2 seconds ago'
>>> timesince(now - datetime.timedelta(seconds=60))
'1 minute ago'
>>> timesince(now - datetime.timedelta(seconds=61))
'1 minute and 1 second ago'
>>> timesince(now - datetime.timedelta(seconds=62))
'1 minute and 2 seconds ago'
>>> timesince(now - datetime.timedelta(seconds=120))
'2 minutes ago'
>>> timesince(now - datetime.timedelta(seconds=121))
'2 minutes and 1 second ago'
>>> timesince(now - datetime.timedelta(seconds=122))
'2 minutes and 2 seconds ago'
>>> timesince(now - datetime.timedelta(seconds=3599))
'59 minutes and 59 seconds ago'
>>> timesince(now - datetime.timedelta(seconds=3600))
'1 hour ago'
>>> timesince(now - datetime.timedelta(seconds=3601))
'1 hour and 1 second ago'
>>> timesince(now - datetime.timedelta(seconds=3602))
'1 hour and 2 seconds ago'
>>> timesince(now - datetime.timedelta(seconds=3660))
'1 hour and 1 minute ago'
>>> timesince(now - datetime.timedelta(seconds=3661))
'1 hour and 1 minute ago'
>>> timesince(now - datetime.timedelta(seconds=3720))
'1 hour and 2 minutes ago'
>>> timesince(now - datetime.timedelta(seconds=3721))
'1 hour and 2 minutes ago'
>>> timesince(datetime.timedelta(seconds=3721))
'1 hour and 2 minutes ago'
Parameters:
Return type:

str

python_utils.generators module¶

This module provides generator utilities for batching items from iterables and async iterables.

Functions:
abatcher(generator, batch_size=None, interval=None):

Asyncio generator wrapper that returns items with a given batch size or interval (whichever is reached first).

batcher(iterable, batch_size=10):

Generator wrapper that returns items with a given batch size.

async python_utils.generators.abatcher(generator, batch_size=None, interval=None)[source]¶

Asyncio generator wrapper that returns items with a given batch size or interval (whichever is reached first).

Parameters:

  • generator (AsyncGenerator[_T, None] | AsyncIterator[_T]) – The async generator or iterator to batch.

  • batch_size (Optional[int], optional) – The number of items per batch. Defaults to None.

  • interval (Optional[_aliases.delta_type], optional) – The time interval to wait before yielding a batch. Defaults to None.

Yields:

collections.abc.AsyncGenerator[list[_T], None] – A generator that yields batches of items.

Return type:

AsyncGenerator[list[_T], None]

python_utils.generators.batcher(iterable, batch_size=10)[source]¶

Generator wrapper that returns items with a given batch size.

Parameters:

  • iterable (collections.abc.Iterable[_T]) – The iterable to batch.

  • batch_size (int, optional) – The number of items per batch. Defaults to 10.

Yields:

collections.abc.Generator[list[_T], None, None] –

A generator that

yields batches of items.

Return type:

Generator[list[_T], None, None]

python_utils.import_ module¶

This module provides utilities for importing modules and handling exceptions.

Classes:
DummyError(Exception):

A custom exception class used as a default for exception handling.

Functions:
import_global(name, modules=None, exceptions=DummyError, locals_=None,

globals_=None, level=-1): Imports the requested items into the global scope, with support for relative imports and custom exception handling.

exception python_utils.import_.DummyError[source]¶

Bases: Exception

A custom exception class used as a default for exception handling.

python_utils.import_.DummyException¶

Backwards-compatible legacy alias for DummyError.

python_utils.import_.import_global(name, modules=None, exceptions=<class 'python_utils.import_.DummyError'>, locals_=None, globals_=None, level=-1)[source]¶

Import the requested items into the global scope.

WARNING! this method _will_ overwrite your global scope If you have a variable named path and you call import_global(‘sys’) it will be overwritten with sys.path

Parameters:

  • name (str) – the name of the module to import, e.g. sys

  • modules (str) – the modules to import, use None for everything

  • exceptions (Exception) – the exception to catch, e.g. ImportError

  • locals – the locals() method (in case you need a different scope)

  • globals – the globals() method (in case you need a different scope)

  • level (int) – the level to import from, this can be used for

  • imports (relative)

  • locals_ (dict[str, Any] | None)

  • globals_ (dict[str, Any] | None)

Return type:

Any

python_utils.logger module¶

This module provides a base class LoggerBase and a derived class Logged for adding logging capabilities to classes. The LoggerBase class expects a logger attribute to be a logging.Logger or compatible instance and provides methods for logging at various levels. The Logged class automatically adds a named logger to the class.

Classes:
LoggerBase:

A base class that adds logging utilities to a class.

Logged:

A derived class that automatically adds a named logger to a class.

Example

>>> class MyClass(Logged):
...     def __init__(self):
...         Logged.__init__(self)

>>> my_class = MyClass()
>>> my_class.debug('debug')
>>> my_class.info('info')
>>> my_class.warning('warning')
>>> my_class.error('error')
>>> my_class.exception('exception')
>>> my_class.log(0, 'log')
class python_utils.logger.Logged(*args, **kwargs)[source]¶

Bases: LoggerBase

Class which automatically adds a named logger to your class when inheriting.

Adds easy access to debug, info, warning, error, exception and log methods

>>> class MyClass(Logged):
...     def __init__(self):
...         Logged.__init__(self)

>>> my_class = MyClass()
>>> my_class.debug('debug')
>>> my_class.info('info')
>>> my_class.warning('warning')
>>> my_class.error('error')
>>> my_class.exception('exception')
>>> my_class.log(0, 'log')

>>> my_class._Logged__get_name('spam')
'spam'
Parameters:
Return type:

typing_extensions.Self

logger: Logger¶

python_utils.loguru module¶

This module provides a Logurud class that integrates the loguru logger with the base logging functionality defined in logger_module.LoggerBase.

Classes:

Logurud: A class that extends LoggerBase and uses loguru for logging.

Usage example:
>>> from python_utils.loguru import Logurud
>>> class MyClass(Logurud):
...     def __init__(self):
...         Logurud.__init__(self)
>>> my_class = MyClass()
>>> my_class.logger.info('This is an info message')
class python_utils.loguru.Logurud(*args, **kwargs)[source]¶

Bases: LoggerBase

A class that extends LoggerBase and uses loguru for logging.

Parameters:
Return type:

Logurud

logger¶

The loguru logger instance.

Type:

loguru.Logger

python_utils.terminal module¶

This module provides functions to get the terminal size across different platforms.

Functions:

get_terminal_size: Get the current size of the terminal. _get_terminal_size_windows: Get terminal size on Windows. _get_terminal_size_tput: Get terminal size using tput. _get_terminal_size_linux: Get terminal size on Linux.

Usage example:
>>> width, height = get_terminal_size()
python_utils.terminal.Dimensions¶

A terminal size as (width, height) in character cells.

alias of tuple[int, int]

python_utils.terminal.OptionalDimensions = tuple[int, int] | None¶

A Dimensions tuple, or None if the size could not be determined.

python_utils.terminal.get_terminal_size()[source]¶

Get the current size of your terminal.

Multiple returns are not always a good idea, but in this case it greatly simplifies the code so I believe it’s justified. It’s not the prettiest function but that’s never really possible with cross-platform code.

Returns:

Two integers containing width and height

Return type:

width, height

python_utils.time module¶

This module provides utility functions for handling time-related operations.

Functions:

timedelta_to_seconds: Convert a timedelta to seconds (microseconds as
    fraction).
delta_to_seconds: Convert a timedelta or numeric interval to seconds.
delta_to_seconds_or_none: Convert a timedelta to seconds or return None.
format_time: Format a timestamp (timedelta, datetime, or seconds).
timeout_generator: Generate items from an iterable until a timeout.
aio_timeout_generator: Async generate items from an iterable until a
    timeout.
aio_generator_timeout_detector: Detect if an async generator has stalled.
aio_generator_timeout_detector_decorator: Decorator for the detector.
async python_utils.time.aio_generator_timeout_detector(generator, timeout=None, total_timeout=None, on_timeout=<function reraise>, **on_timeout_kwargs)[source]¶

This function is used to detect if an asyncio generator has not yielded an element for a set amount of time.

The on_timeout argument is called with the generator, timeout, total_timeout, exception and the extra **kwargs to this function as arguments. If on_timeout is not specified, the exception is reraised. If on_timeout is None, the exception is silently ignored and the generator will finish as normal.

Parameters:
Return type:

AsyncGenerator[_T, None]

python_utils.time.aio_generator_timeout_detector_decorator(timeout=None, total_timeout=None, on_timeout=<function reraise>, **on_timeout_kwargs)[source]¶

Wrap a generator function with aio_generator_timeout_detector.

Parameters:
Returns:

A decorator that wraps an async-generator function so every call is guarded against stalls.

Return type:

Callable[[Callable[[~_P], AsyncGenerator[_T, None]]], Callable[[~_P], AsyncGenerator[_T, None]]]

async python_utils.time.aio_timeout_generator(timeout, interval=datetime.timedelta(seconds=1), iterable=None, interval_multiplier=1.0, maximum_interval=None)[source]¶

Async generator that walks through the given async iterable (a counter by default) until the float_timeout is reached with a configurable float_interval between items.

The interval_exponent automatically increases the float_timeout with each run. Note that if the float_interval is less than 1, 1/interval_exponent will be used so the float_interval is always growing. To double the float_interval with each run, specify 2.

Doctests and asyncio are not friends, so no examples. But this function is effectively the same as the timeout_generator but it uses async for instead.

Parameters:
Return type:

AsyncGenerator[_T, None]

python_utils.time.delta_to_seconds(interval)[source]¶

Convert a timedelta to seconds.

>>> delta_to_seconds(datetime.timedelta(seconds=1))
1
>>> delta_to_seconds(datetime.timedelta(seconds=1, microseconds=1))
1.000001
>>> delta_to_seconds(1)
1
>>> delta_to_seconds('whatever')
Traceback (most recent call last):
    ...
TypeError: Unknown type ...
Parameters:

interval (timedelta | int | float)

Return type:

int | float

python_utils.time.delta_to_seconds_or_none(interval)[source]¶

Convert a timedelta to seconds, passing None through unchanged.

Parameters:

interval (timedelta | int | float | None) – A timedelta or a number of seconds, or None.

Returns:

The interval in seconds, or None when interval is None.

Return type:

int | float | None

>>> delta_to_seconds_or_none(datetime.timedelta(seconds=2))
2
>>> delta_to_seconds_or_none(None) is None
True
python_utils.time.format_time(timestamp, precision=datetime.timedelta(seconds=1))[source]¶

Formats timedelta/datetime/seconds.

>>> format_time('1')
'0:00:01'
>>> format_time(1.234)
'0:00:01'
>>> format_time(1)
'0:00:01'
>>> format_time(datetime.datetime(2000, 1, 2, 3, 4, 5, 6))
'2000-01-02 03:04:05'
>>> format_time(datetime.date(2000, 1, 2))
'2000-01-02'
>>> format_time(datetime.timedelta(seconds=3661))
'1:01:01'
>>> format_time(None)
'--:--:--'
>>> format_time(format_time)
Traceback (most recent call last):
    ...
TypeError: Unknown type ...
Parameters:
Return type:

str

python_utils.time.timedelta_to_seconds(delta)[source]¶

Convert a timedelta to seconds with the microseconds as fraction.

Note that this method has become largely obsolete with the timedelta.total_seconds() method introduced in Python 2.7.

>>> from datetime import timedelta
>>> '%d' % timedelta_to_seconds(timedelta(days=1))
'86400'
>>> '%d' % timedelta_to_seconds(timedelta(seconds=1))
'1'
>>> '%.6f' % timedelta_to_seconds(timedelta(seconds=1, microseconds=1))
'1.000001'
>>> '%.6f' % timedelta_to_seconds(timedelta(microseconds=1))
'0.000001'
Parameters:

delta (timedelta)

Return type:

int | float

python_utils.time.timeout_generator(timeout, interval=datetime.timedelta(seconds=1), iterable=<class 'itertools.count'>, interval_multiplier=1.0, maximum_interval=None)[source]¶

Generator that walks through the given iterable (a counter by default) until the float_timeout is reached with a configurable float_interval between items.

This can be used to limit the time spent on a slow operation. This can be useful for testing slow APIs so you get a small sample of the data in a reasonable amount of time.

>>> for i in timeout_generator(0.1, 0.06):
...     # Put your slow code here
...     print(i)
0
1
2
>>> timeout = datetime.timedelta(seconds=0.1)
>>> interval = datetime.timedelta(seconds=0.06)
>>> for i in timeout_generator(timeout, interval, itertools.count()):
...     print(i)
0
1
2
>>> for i in timeout_generator(1, interval=0.1, iterable='ab'):
...     print(i)
a
b

>>> timeout = datetime.timedelta(seconds=0.1)
>>> interval = datetime.timedelta(seconds=0.06)
>>> for i in timeout_generator(timeout, interval, interval_multiplier=2):
...     print(i)
0
1
2
Parameters:
Return type:

Iterable[_T]

python_utils.types module¶

This module provides type definitions and utility functions for type hinting.

It includes: - Shorthand for commonly used types such as Optional and Union. - Type aliases for various data structures and common types. - Importing all types from the typing and typing_extensions modules. - Importing specific types from the types module.

The module also configures Pyright to ignore wildcard import warnings.

Note

python_utils.types re-exports everything from typing, typing_extensions and types, plus the lightweight aliases below (defined in a stdlib-only module so importing them stays cheap): Scope, OptionalScope, Number, DecimalNumber, ExceptionType, ExceptionsType, StringTypes, delta_type and timestamp_type. The shorthands O (Optional) and U (Union) are kept for backwards compatibility.

Module contents¶

This module initializes the python_utils package by importing various submodules and functions.

Imports are performed lazily (PEP 562): nothing is imported when you import python_utils; each submodule/function is loaded on first access. This keeps import python_utils cheap and, in particular, avoids eagerly importing asyncio (via the async helpers) for consumers that only need the synchronous utilities.

Submodules:

aio
converters
decorators
formatters
generators
import_
logger
terminal
time
types

Functions:

acount
remap
scale_1024
to_float
to_int
to_str
to_unicode
listify
set_attributes
raise_exception
reraise
camel_to_underscore
timesince
abatcher
batcher
import_global
get_terminal_size
aio_generator_timeout_detector
aio_generator_timeout_detector_decorator
aio_timeout_generator
delta_to_seconds
delta_to_seconds_or_none
format_time
timedelta_to_seconds
timeout_generator

Classes:

CastedDict
LazyCastedDict
UniqueList
Logged
LoggerBase