Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@yappi.profile(return_callback=aggregate)
def a(x, y):
if x + y == 25:
raise Exception("")
return x + y
@yappi.profile(return_callback=aggregate)
def count_down_rec(n):
if n == 0:
return
count_down_rec(n - 1)
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_storage_mixin(self):
puts("Bootstrapping storage...")
try:
self.bootstrap_storage()
# Make sure settings are there.
self.storage.verify_settings()
with indent(2):
show_valid("Bootstrapped!")
puts("")
except ImportError :
module_name = traceback.format_exc().split(" ")[-1]
error("Unable to bootstrap storage - attempting to load %s" % module_name)
puts(traceback.format_exc())
sys.exit(1)
except Exception:
error("Unable to bootstrap storage!")
@yappi_profile(return_callback=yappi_aggregate)
def verify_execution(self):
puts("Verifying Execution backend...")
missing_settings = False
missing_setting_error_messages = []
one_valid_backend = False
if not hasattr(settings, "EXECUTION_BACKENDS"):
settings.EXECUTION_BACKENDS = ["will.backends.execution.all", ]
with indent(2):
for b in settings.EXECUTION_BACKENDS:
try:
path_name = None
for mod in b.split('.'):
if path_name is not None:
path_name = [path_name]
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_analysis(self):
self.analysis_backends = []
self.analysis_threads = []
for b in settings.ANALYZE_BACKENDS:
module = import_module(b)
for class_name, cls in inspect.getmembers(module, predicate=inspect.isclass):
try:
if (
hasattr(cls, "is_will_analysisbackend") and
cls.is_will_analysisbackend and
class_name != "AnalysisBackend"
):
c = cls()
thread = Process(
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_io(self):
# puts("Bootstrapping IO...")
self.has_stdin_io_backend = False
self.io_backends = []
self.io_threads = []
self.stdin_io_backends = []
for b in self.valid_io_backends:
module = import_module(b)
for class_name, cls in inspect.getmembers(module, predicate=inspect.isclass):
try:
if (
hasattr(cls, "is_will_iobackend") and
cls.is_will_iobackend and
class_name != "IOBackend" and
class_name != "StdInOutIOBackend"
):
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_bottle(self):
bootstrapped = False
try:
for cls, function_name in self.bottle_routes:
instantiated_cls = cls(bot=self)
instantiated_fn = getattr(instantiated_cls, function_name)
bottle_route_args = {}
for k, v in instantiated_fn.will_fn_metadata.items():
if "bottle_" in k and k != "bottle_route":
bottle_route_args[k[len("bottle_"):]] = v
bottle.route(instantiated_fn.will_fn_metadata["bottle_route"], **bottle_route_args)(instantiated_fn)
bootstrapped = True
except Exception as e:
self.startup_error("Error bootstrapping bottle", e)
if bootstrapped:
show_valid("Web server started at %s." % (settings.PUBLIC_URL,))
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_generation(self):
self.generation_backends = []
self.generation_threads = []
for b in settings.GENERATION_BACKENDS:
module = import_module(b)
for class_name, cls in inspect.getmembers(module, predicate=inspect.isclass):
try:
if (
hasattr(cls, "is_will_generationbackend") and
cls.is_will_generationbackend and
class_name != "GenerationBackend"
):
c = cls()
thread = Process(
target=c.start,
@yappi_profile(return_callback=yappi_aggregate)
def bootstrap_event_handler(self):
self.analysis_timeout = getattr(settings, "ANALYSIS_TIMEOUT_MS", 2000)
self.generation_timeout = getattr(settings, "GENERATION_TIMEOUT_MS", 2000)
self.pubsub.subscribe(["message.*", "analysis.*", "generation.*"])
# TODO: change this to the number of running analysis threads
num_analysis_threads = len(settings.ANALYZE_BACKENDS)
num_generation_threads = len(settings.GENERATION_BACKENDS)
analysis_threads = {}
generation_threads = {}
while True:
try:
event = self.pubsub.get_message()
if event and hasattr(event, "type"):
now = datetime.datetime.now()