iamai.utils module

A utility used internally by iamai.

class iamai.utils.ModulePathFinder[source]

Bases: MetaPathFinder

Meta path finder for finding iamai components.

find_spec(fullname: str, path: Sequence[str] | None = None, target: ModuleType | None = None) → ModuleSpec | None[source]

Used to find the spec of a specified module.

path: ClassVar[List[str]] = []
class iamai.utils.PydanticEncoder(*, skipkeys=False, ensure_ascii=True, check_circular=True, allow_nan=True, sort_keys=False, indent=None, separators=None, default=None)[source]

Bases: JSONEncoder

JSONEncoder class for parsing pydantic.BaseModel.

default(o: Any) → Any[source]

Returns a serializable object of o.

iamai.utils.get_annotations(obj, *, globals=None, locals=None, eval_str=False)[source]

Compute the annotations dict for an object.

obj may be a callable, class, or module. Passing in an object of any other type raises TypeError.

Returns a dict. get_annotations() returns a new dict every time it’s called; calling it twice on the same object will return two different but equivalent dicts.

This function handles several details for you:

  • If eval_str is true, values of type str will be un-stringized using eval(). This is intended for use with stringized annotations (“from __future__ import annotations”).

  • If obj doesn’t have an annotations dict, returns an empty dict. (Functions and methods always have an annotations dict; classes, modules, and other types of callables may not.)

  • Ignores inherited annotations on classes. If a class doesn’t have its own annotations dict, returns an empty dict.

  • All accesses to object members and dict values are done using getattr() and dict.get() for safety.

  • Always, always, always returns a freshly-created dict.

eval_str controls whether or not values of type str are replaced with the result of calling eval() on those values:

  • If eval_str is true, eval() is called on values of type str.

  • If eval_str is false (the default), values of type str are unchanged.

globals and locals are passed in to eval(); see the documentation for eval() for more information. If either globals or locals is None, this function may replace that value with a context-specific default, contingent on type(obj):

  • If obj is a module, globals defaults to obj.__dict__.

  • If obj is a class, globals defaults to sys.modules[obj.__module__].__dict__ and locals defaults to the obj class namespace.

  • If obj is a callable, globals defaults to obj.__globals__, although if obj is a wrapped function (using functools.update_wrapper()) it is first unwrapped.

iamai.utils.get_classes_from_module(module: ModuleType, super_class: _TypeT) → List[_TypeT][source]

Find a class of the specified type from the module.

Parameters:

  • module – Python module.

  • super_class – The superclass of the class to be found.

Returns:

Returns a list of classes that meet the criteria.

iamai.utils.get_classes_from_module_name(name: str, super_class: _TypeT, *, reload: bool = False) → List[Tuple[_TypeT, ModuleType]][source]

Find a class of the specified type from the module with the specified name.

Parameters:

  • name – module name, the format is the same as the Python import statement.

  • super_class – The superclass of the class to be found.

  • reload – Whether to reload the module.

Returns:

Returns a list of tuples consisting of classes and modules that meet the criteria.

Raises:

ImportError – An error occurred while importing the module.

iamai.utils.is_config_class(config_class: Any) → TypeGuard[Type[ConfigModel]][source]

Determine whether an object is a configuration class.

Parameters:

config_class – The object to be judged.

Returns:

Returns whether it is a configuration class.

iamai.utils.samefile(path1: str | bytes | PathLike[str] | PathLike[bytes], path2: str | bytes | PathLike[str] | PathLike[bytes]) → bool[source]

A simple wrapper around os.path.samefile.

Parameters:

  • path1 – path1.

  • path2 – path 2.

Returns:

If two paths point to the same file or directory.

iamai.utils.sync_ctx_manager_wrapper(cm: ContextManager[_T], *, to_thread: bool = False) → AsyncGenerator[_T, None][source]

Wrap a synchronous context manager into an asynchronous context manager.

Parameters:

  • cm – The synchronization context manager to be wrapped.

  • to_thread – Whether to run the synchronization function in a separate thread. Defaults to False.

Returns:

Asynchronous context manager.

iamai.utils.sync_func_wrapper(func: Callable[[_P], _R], *, to_thread: bool = False) → Callable[[_P], Coroutine[None, None, _R]][source]

Wrap a synchronous function as an asynchronous function.

Parameters:

  • func – synchronous function to be packaged.

  • to_thread – Whether to run the synchronization function in a separate thread. Defaults to False.

Returns:

Asynchronous functions.

iamai.utils.wrap_get_func(func: Callable[[EventT], bool | Awaitable[bool]] | None) → Callable[[EventT], Awaitable[bool]][source]

Wrap the parameters accepted by the get() function into an asynchronous function.

Parameters:

func – The parameters accepted by the get() function.

Returns:

Asynchronous functions.