Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
template_environment.filters['draw_progressbar'] = draw_progressbar
template_environment.filters['get_operation_status'] = get_operation_status
template_environment.filters['job_filter'] = job_filter
template = template_environment.get_template(template)
context = self._get_standard_template_context()
# get job status information
if profile:
try:
import pprofile
except ImportError:
raise RuntimeWarning(
"Profiling requires the pprofile package. "
"Install with `pip install pprofile`.")
prof = pprofile.StatisticalProfile()
fn_filter = [
inspect.getfile(threading),
inspect.getfile(multiprocessing),
inspect.getfile(Pool),
inspect.getfile(ThreadPool),
inspect.getfile(tqdm),
]
with prof(single=False):
tmp = self._fetch_status(jobs, err, ignore_errors, no_parallelize)
prof._mergeFileTiming()
# Unrestricted
total_impact = 0
def start_profiling(func, filepath, statistical=True):
import pprofile
if statistical:
prof = pprofile.StatisticalProfile()
else:
prof = pprofile.Profile()
def stop_profiling(prof, filepath):
print('Writing profiling data: %s' % filepath, file=sys.stderr)
print('You can use kcachegrind to analyze it.', file=sys.stderr)
with open(filepath, 'w') as f:
prof.callgrind(f)
# This makes the `finally` block work as expected if we're terminated by
# SIGTERM, which happens by default when running `timeout 30 stig ...`.
# https://stackoverflow.com/a/42200623
# https://mail.python.org/pipermail/python-ideas/2016-February/038474.html
import signal
class SigTerm(SystemExit): pass
def sigterm(sig, frame): raise SigTerm
(the rest)
"""
out = BytesIO()
with zipfile.ZipFile(
out,
mode='w',
compression=zipfile.ZIP_DEFLATED,
) as outfile:
for path, data, _ in self._iterOutFiles():
outfile.writestr(path, data)
return out.getvalue(), 'application/zip'
class ZopeProfiler(ZopeMixIn, pprofile.Profile):
__slots__ = ZopeMixIn.virtual__slots__
class ZopeStatisticalProfile(ZopeMixIn, pprofile.StatisticalProfile):
__slots__ = ZopeMixIn.virtual__slots__
class ZopeStatisticalThread(pprofile.StatisticalThread):
__allow_access_to_unprotected_subobjects__ = 1
# Intercept "verbose" parameter to prevent writing to stdout.
def getProfiler(verbose=False, **kw):
"""
Get a Zope-friendly pprofile.Profile instance.
"""
return ZopeProfiler(**kw)
def getStatisticalProfilerAndThread(**kw):
"""
Get Zope-friendly pprofile.StatisticalProfile and
pprofile.StatisticalThread instances.