Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# so the name was missing etc, just proxy into original open call and let it puke
raise _EarlyExit("no name/file was given")
file = kwargs.get(filearg)
if isinstance(file, int):
raise _EarlyExit("already a file descriptor")
if self._paths_cache is not None:
filefull = file if isabs(file) else os.path.abspath(file)
if filefull in self._paths_cache:
raise _EarlyExit("considered before")
else:
self._paths_cache.add(filefull)
if _DOT_GIT_DIR in file:
raise _EarlyExit("we ignore paths under .git/")
mode = 'r'
if len(args) > 1:
mode = args[1]
elif 'mode' in kwargs:
mode = kwargs['mode']
if 'r' in mode:
self._dataset_auto_get(file)
else:
raise _EarlyExit("mode=%r", mode)
except _EarlyExit as e:
lgr.log(2, " skipping since " + e.msg, *e.args,
extra={'notraceback': True})
except Exception as e:
# If anything goes wrong -- we should complain and proceed
with self._patch(origname, origfunc):
def _proxy_open_name_mode(self, origname, origfunc, *args, **kwargs):
"""Proxy for various "open" which have first argument name and 2nd - mode
"""
# wrap it all for resilience to errors -- proxying must do no harm!
try:
if self._in_open:
raise _EarlyExit("within open already")
self._in_open = True # just in case someone kept alias/assignment
# return stock open for the duration of handling so that
# logging etc could workout correctly
with self._patch(origname, origfunc):
lgr.log(3, "Proxying open with %r %r", args, kwargs)
# had to go with *args since in PY2 it is name, in PY3 file
# deduce arguments
if len(args) > 0:
# name/file was provided
file = args[0]
else:
filearg = "name" if PY2 else "file"
if filearg not in kwargs:
# so the name was missing etc, just proxy into original open call and let it puke
raise _EarlyExit("no name/file was given")
lgr.log(3, "Proxying open with %r %r", args, kwargs)
# had to go with *args since in PY2 it is name, in PY3 file
# deduce arguments
if len(args) > 0:
# name/file was provided
file = args[0]
else:
filearg = "name" if PY2 else "file"
if filearg not in kwargs:
# so the name was missing etc, just proxy into original open call and let it puke
raise _EarlyExit("no name/file was given")
file = kwargs.get(filearg)
if isinstance(file, int):
raise _EarlyExit("already a file descriptor")
if self._paths_cache is not None:
filefull = file if isabs(file) else os.path.abspath(file)
if filefull in self._paths_cache:
raise _EarlyExit("considered before")
else:
self._paths_cache.add(filefull)
if _DOT_GIT_DIR in file:
raise _EarlyExit("we ignore paths under .git/")
mode = 'r'
if len(args) > 1:
mode = args[1]
elif 'mode' in kwargs:
mode = kwargs['mode']
# name/file was provided
file = args[0]
else:
filearg = "name" if PY2 else "file"
if filearg not in kwargs:
# so the name was missing etc, just proxy into original open call and let it puke
raise _EarlyExit("no name/file was given")
file = kwargs.get(filearg)
if isinstance(file, int):
raise _EarlyExit("already a file descriptor")
if self._paths_cache is not None:
filefull = file if isabs(file) else os.path.abspath(file)
if filefull in self._paths_cache:
raise _EarlyExit("considered before")
else:
self._paths_cache.add(filefull)
if _DOT_GIT_DIR in file:
raise _EarlyExit("we ignore paths under .git/")
mode = 'r'
if len(args) > 1:
mode = args[1]
elif 'mode' in kwargs:
mode = kwargs['mode']
if 'r' in mode:
self._dataset_auto_get(file)
else:
raise _EarlyExit("mode=%r", mode)
except _EarlyExit as e:
raise _EarlyExit("considered before")
else:
self._paths_cache.add(filefull)
if _DOT_GIT_DIR in file:
raise _EarlyExit("we ignore paths under .git/")
mode = 'r'
if len(args) > 1:
mode = args[1]
elif 'mode' in kwargs:
mode = kwargs['mode']
if 'r' in mode:
self._dataset_auto_get(file)
else:
raise _EarlyExit("mode=%r", mode)
except _EarlyExit as e:
lgr.log(2, " skipping since " + e.msg, *e.args,
extra={'notraceback': True})
except Exception as e:
# If anything goes wrong -- we should complain and proceed
with self._patch(origname, origfunc):
lgr.warning("Failed proxying open with %r, %r: %s", args, kwargs, exc_str(e))
finally:
self._in_open = False
# finally give it back to stock open
return origfunc(*args, **kwargs)
self._in_open = True # just in case someone kept alias/assignment
# return stock open for the duration of handling so that
# logging etc could workout correctly
with self._patch(origname, origfunc):
lgr.log(3, "Proxying open with %r %r", args, kwargs)
# had to go with *args since in PY2 it is name, in PY3 file
# deduce arguments
if len(args) > 0:
# name/file was provided
file = args[0]
else:
filearg = "name" if PY2 else "file"
if filearg not in kwargs:
# so the name was missing etc, just proxy into original open call and let it puke
raise _EarlyExit("no name/file was given")
file = kwargs.get(filearg)
if isinstance(file, int):
raise _EarlyExit("already a file descriptor")
if self._paths_cache is not None:
filefull = file if isabs(file) else os.path.abspath(file)
if filefull in self._paths_cache:
raise _EarlyExit("considered before")
else:
self._paths_cache.add(filefull)
if _DOT_GIT_DIR in file:
raise _EarlyExit("we ignore paths under .git/")
mode = 'r'
if len(args) > 1: