iamai.utils module#

iamai 内部使用的实用工具。

class iamai.utils.ModulePathFinder[source]#

Bases: MetaPathFinder

用于查找 iamai 组件的元路径查找器。

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

用于查找指定模块的 spec

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

用于解析 pydantic.BaseModelJSONEncoder 类。

default(o: Any) Any[source]#

返回 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: module, super_class: _TypeT) List[_TypeT][source]#

从模块中查找指定类型的类。

Parameters:
  • module – Python 模块。

  • super_class – 要查找的类的超类。

Returns:

返回符合条件的类的列表。

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

从指定名称的模块中查找指定类型的类。

Parameters:
  • name – 模块名称,格式和 Python import 语句相同。

  • super_class – 要查找的类的超类。

  • reload – 是否重新加载模块。

Returns:

返回由符合条件的类和模块组成的元组的列表。

Raises:

ImportError – 当导入模块过程中出现错误。

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

判断一个对象是否是配置类。

Parameters:

config_class – 待判断的对象。

Returns:

返回是否是配置类。

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

一个 os.path.samefile 的简单包装。

Parameters:
  • path1 – 路径1。

  • path2 – 路径2。

Returns:

如果两个路径是否指向相同的文件或目录。

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

将同步上下文管理器包装为异步上下文管理器。

Parameters:
  • cm – 待包装的同步上下文管理器。

  • to_thread – 是否在独立的线程中运行同步函数。默认为 False

Returns:

异步上下文管理器。

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

包装一个同步函数为异步函数。

Parameters:
  • func – 待包装的同步函数。

  • to_thread – 是否在独立的线程中运行同步函数。默认为 False

Returns:

异步函数。

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

get() 函数接受的参数包装为一个异步函数。

Parameters:

funcget() 函数接受的参数。

Returns:

异步函数。