Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _handle_fresh_imports(self, name, import_level_prefix, level):
"""Check which modules were imported since last point we checked and add them to the queue
"""
new_imported_modules = set(sys.modules.keys()) - self._processed_modules - self.__queue_to_process
if new_imported_modules:
lgr.log(4, "%s%d new modules were detected upon import of %s (level=%s)",
import_level_prefix, len(new_imported_modules), name, level)
# lgr.log(2, "%s%d new modules were detected: %s, upon import of %s (level=%s)",
# import_level_prefix, len(new_imported_modules), new_imported_modules, name, level)
for imported_mod in new_imported_modules:
if imported_mod in self.__queue_to_process:
# we saw it already
continue
# lgr.log(1, "Name %r was imported as %r (path: %s). fromlist: %s, level: %s",
# name, mod.__name__, getattr(mod, '__path__', None), fromlist, level)
# package
package = imported_mod.split('.', 1)[0]
if package != imported_mod \
and package not in self._processed_modules \
and package not in self.__queue_to_process:
# if its parent package wasn't yet imported before
lgr.log(3, "%sParent of %s, %s wasn't yet processed, adding to the queue",
def _process_delayed_injection(self, mod_name):
lgr.debug("%sProcessing delayed injection for %s", self._import_level_prefix, mod_name)
inj_mod_name = self._delayed_injections[mod_name]
assert(not hasattr(self._orig_import, '__duecredited__'))
try:
inj_mod_name_full = "duecredit.injections." + inj_mod_name
lgr.log(3, "Importing %s", inj_mod_name_full)
# Mark it is a processed already, to avoid its processing etc
self._processed_modules.add(inj_mod_name_full)
inj_mod = self._orig_import(inj_mod_name_full,
fromlist=["duecredit.injections"])
except Exception as e:
if os.environ.get('DUECREDIT_ALLOW_FAIL', False):
raise
raise RuntimeError("Failed to import %s: %r" % (inj_mod_name, e))
# TODO: process min/max_versions etc
assert(hasattr(inj_mod, 'inject'))
lgr.log(3, "Calling injector of %s", inj_mod_name_full)
inj_mod.inject(self)
# go through the known entries and register them within the collector, and
# decorate corresponding methods
# There could be multiple records per module
for obj_path, obj_entry_records in iteritems(self._entry_records[mod_name]):
parent, obj_name = None, None
if obj_path:
# so we point to an object within the mod
try:
parent, obj_name, obj = find_object(mod, obj_path)
except (KeyError, AttributeError) as e:
lgr.warning("Could not find %s in module %s: %s" % (obj_path, mod, e))
continue
# there could be multiple per func
lgr.log(4, "Considering %d records for decoration of %s:%s", len(obj_entry_records), parent, obj_name)
for obj_entry_record in obj_entry_records:
entry = obj_entry_record['entry']
# Add entry explicitly
self._collector.add(entry)
if obj_path: # if not entire module -- decorate!
decorator = self._collector.dcite(entry.get_key(), **obj_entry_record['kwargs'])
lgr.debug("Decorating %s:%s with %s", parent, obj_name, decorator)
obj_decorated = decorator(obj)
setattr(parent, obj_name, obj_decorated)
# override previous obj with the decorated one if there are multiple decorators
obj = obj_decorated
else:
lgr.log(3, "Citing directly %s:%s since obj_path is empty", parent, obj_name)
self._collector.cite(entry.get_key(), **obj_entry_record['kwargs'])
lgr.log(3, "Done processing injections for module %s", mod_name)
for obj_entry_record in obj_entry_records:
entry = obj_entry_record['entry']
# Add entry explicitly
self._collector.add(entry)
if obj_path: # if not entire module -- decorate!
decorator = self._collector.dcite(entry.get_key(), **obj_entry_record['kwargs'])
lgr.debug("Decorating %s:%s with %s", parent, obj_name, decorator)
obj_decorated = decorator(obj)
setattr(parent, obj_name, obj_decorated)
# override previous obj with the decorated one if there are multiple decorators
obj = obj_decorated
else:
lgr.log(3, "Citing directly %s:%s since obj_path is empty", parent, obj_name)
self._collector.cite(entry.get_key(), **obj_entry_record['kwargs'])
lgr.log(3, "Done processing injections for module %s", mod_name)
"""Process import of the module, possibly decorating some methods with duecredit entries
"""
assert(self.__import_level == 0) # we should never process while nested within imports
# We need to mark that module as processed EARLY, so we don't try to re-process it
# while doing _process_delayed_injection
self._processed_modules.add(mod_name)
if mod_name in self._delayed_injections:
# should be hit only once, "theoretically" unless I guess reimport is used etc
self._process_delayed_injection(mod_name)
if mod_name not in self._entry_records:
return
total_number_of_citations = sum(map(len, self._entry_records[mod_name].values()))
lgr.log(logging.DEBUG + 5,
"Process %d citation injections for %d objects for module %s",
total_number_of_citations, len(self._entry_records[mod_name]), mod_name)
try:
mod = sys.modules[mod_name]
except KeyError:
lgr.warning("Failed to access module %s among sys.modules" % mod_name)
return
# go through the known entries and register them within the collector, and
# decorate corresponding methods
# There could be multiple records per module
for obj_path, obj_entry_records in iteritems(self._entry_records[mod_name]):
parent, obj_name = None, None
if obj_path:
# so we point to an object within the mod