feathersdk.utils.common Module

Misc helper functions that can be used throughout the project

exception feathersdk.utils.common.CriticalException

Bases: Exception

An exception that should still be raised and crash the program, even when caught for logging purposes.

class feathersdk.utils.common.FeatherThread(target, name=None, args=(), kwargs=None, daemon=True)

Bases: Thread

A custom threading.Thread. Currently can be used to debug print messages.

class feathersdk.utils.common.FeatherThreadLock(name: str, timeout: float = 5.0, pre_acquire_callback: Callable[[], None] | None = None, post_release_callback: Callable[[], None] | None = None)

Bases: object

A custom threading.Lock. Currently can be used to debug print messages. Call set_debug_lock_print() to enable

Parameters:

  • name (str) – The name of the lock.

  • timeout (float) – The timeout in seconds to wait for the lock to be acquired.

  • pre_acquire_callback (Optional[Callable[[], None]]) – A callback to be called before the lock is acquired.

  • post_release_callback (Optional[Callable[[], None]]) – A callback to be called after the lock is released.

is_locked() → bool

Check if the lock is currently locked.

exception feathersdk.utils.common.LockAcquisitionError

Bases: Exception

Raised when a lock cannot be acquired.

feathersdk.utils.common.TimeType

The type that we use to handle time-related operations.

feathersdk.utils.common.annotation_allows_type(annotation: Any, type_to_check: type) → bool

Check if a type annotation allows for the given type.

feathersdk.utils.common.currtime() → float

Get the current time in seconds.

Currently (pun intended), this returns time.monotonic(), which returns time since computer power-on on linux.

feathersdk.utils.common.dont_allow_on_fork(func)

Decorator to prevent a method from being called after fork.

feathersdk.utils.common.exponential_timeout(timeout_seconds: float, func: ~typing.Callable[[~P], ~feathersdk.utils.common.R | None], func_args: ~typing.~P = (), func_kwargs: ~typing.~P = {}, start_delay: float = 0.0001, max_delay: float = 0.01, mult: float = 1.5) → R | None

Perform an exponential timeout.

Will run the given function, sleeping for longer and longer until the function returns a non-None value, or the timeout is reached (in which case a TimeoutError is raised).

Parameters:

  • timeout_seconds – The timeout in seconds.

  • func – The function to execute. Should return None if the operation is not complete, and a non-None result if it is.

  • func_args – Arguments to pass to the function.

  • func_kwargs – Keyword arguments to pass to the function.

  • start_delay – The starting delay in seconds.

  • max_delay – The maximum delay in seconds.

  • mult – Delay is multiplied by this factor after each iteration. (but not allowed to exceed max_delay)

Raises:

TimeoutError – If the function does not return a non-None value before the timeout is reached.

feathersdk.utils.common.listdirfull(dir: str, startswith: str | None = None, endswith: str | None = None) → List[str]

List all files in the given directory, returning a list of full file paths.

Parameters:

  • dir – The directory to list files from.

  • startswith – If provided, only list files that start with this string.

  • endswith – If provided, only list files that end with this string.

feathersdk.utils.common.make_exception_pickleable(cls: Type[TExc]) → Type[TExc]

Decorate an exception class to modify it to be pickleable.

This is useful for exceptions that are raised in multiprocessing Pool calls as they get pickled and sent to the main process to be re-raised. During the unpickling process, the __init__ method is called and passed a single string argument: the full exception message from the original pickled exception.This means if you create a new exception class which doesn’t take any arguments, or takes arguments which are used to build the exception message, you’ll get an error when unpickling due to incorrect number of arguments.

This function modifies the __init__ method. When called, the first argument is expected to be self. For most calls, the class-overridden __init__ method will be called with the passed args/kwargs. However, if the new __init__ method is called with exactly one argument which is positional and a string, then the superclass’s __init__ method will be called with the string argument, which is expected to be the full final exception message.

NOTE: If there is ambiguity in the arguments to the overridden __init__ method (meaning: it is at all possible to pass a single string parameter and the function call would be valid), we may not be able to tell if the argument is the exception message or the normal positional argument, and thus an exception will be raised. In this case, you can either set a non-string type annotation on what you pass in to the __init__ method, or make the argument keyword-only.

Parameters:

cls – The exception class to modify. Should inherit from Exception.

feathersdk.utils.common.progressbar(iterable: Iterable[T], progress: bool = True) → Iterable[T]
feathersdk.utils.common.set_debug_lock_print(debug_lock_print: bool) → None

Set whether to print debug messages when acquiring and releasing locks.

feathersdk.utils.common.timediff(start_time: float) → float

Get the time difference in seconds between the current time and the given start time.

feathersdk.utils.common.timestamped_class(cls: Type[T]) → Type[T]

Decorate a class to add a timestamp attribute to the class.

The timestamp is set to the latest time that: - The __init__ method is called - The __post_init__ method is called (for dataclasses) - The object is created using the __new__ method (in case __init__ or __post_init__ are not called)

feathersdk.utils.common.to_snake_case(name: str) → str

Convert a string to snake case.

feathersdk.utils.common.try_log_exception_peacefully(func: Callable[[P], R] | None = None, *, capture_locals: Iterable[str] | None = None) → Callable[[P], R | None]

Decorate a function to log exceptions.

If an exception is raised, it will be logged and the exception will be ignored, unless the exception is a CriticalException, in which case it will be re-raised. Caught exceptions will return None.

Can be used either like @try_log_exception_peacefully or @try_log_exception_peacefully(args…).

Parameters:

  • func – Function to decorate. If not provided, will return a decorator that can be used to decorate a function.

  • capture_locals – Iterable of local variable names to capture and log.