Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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)
# 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)
def __del__(self):
if lgr:
lgr.debug("%s is asked to be deleted", self)
try:
if self._active:
self.deactivate()
except: # noqa: E722
pass
try:
super(self.__class__, self).__del__()
except: # noqa: E722
pass
def import_doi(doi, sleep=0.5, retries=10):
import requests
cached = get_doi_cache_file(doi)
if exists(cached):
with open(cached) as f:
doi = f.read()
if PY2:
return doi.decode('utf-8')
return doi
# else -- fetch it
headers = {'Accept': 'application/x-bibtex; charset=utf-8'}
url = 'https://doi.org/' + doi
while retries > 0:
lgr.debug("Submitting GET to %s with headers %s", url, headers)
r = requests.get(url, headers=headers)
r.encoding = 'UTF-8'
bibtex = r.text.strip()
if bibtex.startswith('@'):
# no more retries necessary
break
lgr.warning("Failed to obtain bibtex from doi.org, retrying...")
time.sleep(sleep) # give some time to the server
retries -= 1
status_code = r.status_code
if not bibtex.startswith('@'):
raise ValueError('Query %(url)s for BibTeX for a DOI %(doi)s (wrong doi?) has failed. '
'Response code %(status_code)d. '
#'BibTeX response was: %(bibtex)s'
% locals())
if not exists(cached):
def _populate_delayed_injections(self):
self._delayed_injections = {}
for inj_mod_name in get_modules_for_injection():
assert(inj_mod_name.startswith('mod_'))
mod_name = inj_mod_name[4:]
lgr.debug("Adding delayed injection for %s", (mod_name,))
self._delayed_injections[mod_name] = inj_mod_name