Implement string imports for Factory, Callable, Singletons, and Resource

This commit is contained in:
Roman Mogylatov 2022-01-29 16:49:08 -05:00
parent 38ca1cdeed
commit 6ec978d13e
2 changed files with 10227 additions and 9071 deletions

File diff suppressed because it is too large Load Diff

View File

@ -6,6 +6,7 @@ import copy
import errno import errno
import functools import functools
import inspect import inspect
import importlib
import os import os
import re import re
import sys import sys
@ -18,6 +19,11 @@ try:
except ImportError: except ImportError:
contextvars = None contextvars = None
try:
import builtins
except ImportError:
# Python 2.7
import __builtin__ as builtins
try: try:
import asyncio import asyncio
@ -1221,6 +1227,7 @@ cdef class Callable(Provider):
def set_provides(self, provides): def set_provides(self, provides):
"""Set provider provides.""" """Set provider provides."""
provides = _resolve_provides(provides)
if provides and not callable(provides): if provides and not callable(provides):
raise Error( raise Error(
"Provider {0} expected to get callable, got {1} instead".format( "Provider {0} expected to get callable, got {1} instead".format(
@ -1426,9 +1433,10 @@ cdef class Coroutine(Callable):
_is_coroutine = _is_coroutine_marker _is_coroutine = _is_coroutine_marker
def set_provides(self, provides): def set_provides(self, provides):
"""Set provider"s provides.""" """Set provider provides."""
if not asyncio: if not asyncio:
raise Error("Package asyncio is not available") raise Error("Package asyncio is not available")
provides = _resolve_provides(provides)
if provides and not asyncio.iscoroutinefunction(provides): if provides and not asyncio.iscoroutinefunction(provides):
raise Error(f"Provider {_class_qualname(self)} expected to get coroutine function, " raise Error(f"Provider {_class_qualname(self)} expected to get coroutine function, "
f"got {provides} instead") f"got {provides} instead")
@ -2452,6 +2460,7 @@ cdef class Factory(Provider):
def set_provides(self, provides): def set_provides(self, provides):
"""Set provider provides.""" """Set provider provides."""
provides = _resolve_provides(provides)
if (provides if (provides
and self.__class__.provided_type and and self.__class__.provided_type and
not issubclass(provides, self.__class__.provided_type)): not issubclass(provides, self.__class__.provided_type)):
@ -2742,6 +2751,7 @@ cdef class BaseSingleton(Provider):
def set_provides(self, provides): def set_provides(self, provides):
"""Set provider provides.""" """Set provider provides."""
provides = _resolve_provides(provides)
if (provides if (provides
and self.__class__.provided_type and and self.__class__.provided_type and
not issubclass(provides, self.__class__.provided_type)): not issubclass(provides, self.__class__.provided_type)):
@ -3580,6 +3590,7 @@ cdef class Resource(Provider):
def set_provides(self, provides): def set_provides(self, provides):
"""Set provider provides.""" """Set provider provides."""
provides = _resolve_provides(provides)
self.__provides = provides self.__provides = provides
return self return self
@ -4882,6 +4893,45 @@ def isasyncgenfunction(obj):
return False return False
def _resolve_provides(provides):
if provides is None:
return provides
elif not isinstance(provides, str):
return provides
else:
segments = provides.split(".")
member_name = segments[-1]
if len(segments) == 1:
if member_name in dir(builtins):
module = builtins
else:
module = _resolve_calling_module()
return getattr(module, member_name)
module_name = ".".join(segments[:-1])
package_name = None
if module_name.startswith("."):
package_name = _resolve_calling_package_name()
if package_name is None:
raise ImportError("attempted relative import with no known parent package")
module = importlib.import_module(module_name, package=package_name)
return getattr(module, member_name)
def _resolve_calling_module():
stack = inspect.stack()
pre_last_frame = stack[0]
return inspect.getmodule(pre_last_frame[0])
def _resolve_calling_package_name():
module = _resolve_calling_module()
return module.__package__
cpdef _copy_parent(object from_, object to, dict memo): cpdef _copy_parent(object from_, object to, dict memo):
"""Copy and assign provider parent.""" """Copy and assign provider parent."""
copied_parent = ( copied_parent = (