"""Utilities related to importing modules and symbols by name."""importosimportsysimportwarningsfromcontextlibimportcontextmanagerfromimportlibimportimport_module,reloadtry:fromimportlib.metadataimportentry_pointsexceptImportError:fromimportlib_metadataimportentry_pointsfromkombu.utils.importsimportsymbol_by_name#: Billiard sets this when execv is enabled.#: We use it to find out the name of the original ``__main__``#: module, so that we can properly rewrite the name of the#: task to be that of ``App.main``.MP_MAIN_FILE=os.environ.get('MP_MAIN_FILE')__all__=('NotAPackage','qualname','instantiate','symbol_by_name','cwd_in_path','find_module','import_from_cwd','reload_from_cwd','module_file','gen_task_name',)
[文档]classNotAPackage(Exception):"""Raised when importing a package, but it's not a package."""
[文档]definstantiate(name,*args,**kwargs):"""Instantiate class by name. See Also: :func:`symbol_by_name`. """returnsymbol_by_name(name)(*args,**kwargs)
[文档]@contextmanagerdefcwd_in_path():"""Context adding the current working directory to sys.path."""try:cwd=os.getcwd()exceptFileNotFoundError:cwd=Noneifnotcwd:yieldelifcwdinsys.path:yieldelse:sys.path.insert(0,cwd)try:yieldcwdfinally:try:sys.path.remove(cwd)exceptValueError:# pragma: no coverpass
[文档]deffind_module(module,path=None,imp=None):"""Version of :func:`imp.find_module` supporting dots."""ifimpisNone:imp=import_modulewithcwd_in_path():try:returnimp(module)exceptImportError:# Raise a more specific error if the problem is that one of the# dot-separated segments of the module name is not a package.if'.'inmodule:parts=module.split('.')fori,partinenumerate(parts[:-1]):package='.'.join(parts[:i+1])try:mpart=imp(package)exceptImportError:# Break out and re-raise the original ImportError# instead.breaktry:mpart.__path__exceptAttributeError:raiseNotAPackage(package)raise
[文档]defimport_from_cwd(module,imp=None,package=None):"""Import module, temporarily including modules in the current directory. Modules located in the current directory has precedence over modules located in `sys.path`. """ifimpisNone:imp=import_modulewithcwd_in_path():returnimp(module,package=package)
[文档]defreload_from_cwd(module,reloader=None):"""Reload module (ensuring that CWD is in sys.path)."""ifreloaderisNone:reloader=reloadwithcwd_in_path():returnreloader(module)
[文档]defmodule_file(module):"""Return the correct original file name of a module."""name=module.__file__returnname[:-1]ifname.endswith('.pyc')elsename
[文档]defgen_task_name(app,name,module_name):"""Generate task name from name/module pair."""module_name=module_nameor'__main__'try:module=sys.modules[module_name]exceptKeyError:# Fix for manage.py shell_plus (Issue #366)module=NoneifmoduleisnotNone:module_name=module.__name__# - If the task module is used as the __main__ script# - we need to rewrite the module part of the task name# - to match App.main.ifMP_MAIN_FILEandmodule.__file__==MP_MAIN_FILE:# - see comment about :envvar:`MP_MAIN_FILE` above.module_name='__main__'ifmodule_name=='__main__'andapp.main:return'.'.join([app.main,name])return'.'.join(pforpin(module_name,name)ifp)