Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def get_candidates(beer):
"""
returns a list of donuts based on hops level of beer
"""
span = tracer.current_span()
span.set_tags({'beer.name': beer.name, 'beer.hops': beer.hops})
db = DonutStats.instance()
# find our optimal sugar level Donuts above or below this level
# will certainly not be a good match
optimal_sugar_level = db.get_optimal_sugar_level(beer.hops)
return db.get_by_sugar_level(optimal_sugar_level, limit=10)
def best_match(beer):
"""
returns a single donut matched to the hops level of a beer
"""
# get a list of donuts that match sugar content for beer
candidates = get_candidates(beer)
# send the remaining candidates to our taster and pick the best
max_score = -1
best_match = None
for candidate in candidates:
try:
# propagate the trace context between the two services
span = tracer.current_span()
headers = {
"X-Datadog-Trace-Id": str(span.trace_id),
"X-Datadog-Parent-Id": str(span.span_id),
}
resp = requests.get(
"http://taster:5001/taste",
params={"beer": beer.name, "donut": candidate},
timeout=2,
headers=headers,
)
except requests.exceptions.Timeout:
continue
score = resp.json()["score"]
if score > max_score:
def get_candidates(beer):
"""
returns a list of donuts based on hops level of beer
"""
span = tracer.current_span()
span.set_tags({'beer.name': beer.name, 'beer.hops': beer.hops})
db = DonutStats.instance()
# find our optimal sugar level Donuts above or below this level
# will certainly not be a good match
optimal_sugar_level = db.get_optimal_sugar_level(beer.hops)
return db.get_by_sugar_level(optimal_sugar_level, limit=10)
def _get_dd_trace_py_context():
span = tracer.current_span()
if not span:
return None
parent_id = span.context.span_id
trace_id = span.context.trace_id
sampling_priority = span.context.sampling_priority
return {
"parent-id": str(parent_id),
"trace-id": str(trace_id),
"sampling-priority": str(sampling_priority),
"source": TraceContextSource.DDTRACE,
}
def taste():
# continue the trace
trace_id = request.headers.get("X-Datadog-Trace-Id")
parent_id = request.headers.get("X-Datadog-Parent-Id")
if trace_id and parent_id:
span = tracer.current_span()
span.trace_id = int(trace_id)
span.parent_id = int(parent_id)
beer = request.args.get("beer")
score = 10 if beer == 'stout' else random.randint(1, 10)
return jsonify(score=score)
def taste():
# continue the trace
trace_id = request.headers.get("X-Datadog-Trace-Id")
parent_id = request.headers.get("X-Datadog-Parent-Id")
if trace_id and parent_id:
span = tracer.current_span()
span.trace_id = int(trace_id)
span.parent_id = int(parent_id)
# send the remaining candidates to our taster and pick the best
matches = []
beer = request.args.get("beer")
candidates = request.args.getlist("donuts")
for candidate in candidates:
score = 10 if beer == 'stout' else random.randint(1, 10)
matches.append((score, candidate))
best_match = max(matches)
return jsonify(candidate=best_match[1], score=best_match[0])
async def handle_request(reader, writer):
# trace something
with tracer.trace('async.handler', service='asyncio-web') as span:
# do something
await asyncio.sleep(0.02)
# in the meantime do something else
value = await get_redis_value()
# do something that will be conclude in the future
future = helpers.ensure_future(delayed_job(tracer.current_span()))
# response
start_response(writer)
writer.write(b'OK\r\n')
writer.close()
await future
print('200: request handled')
def get_candidates(beer):
"""
returns a list of donuts based on hops level of beer
"""
span = tracer.current_span()
span.set_tags({'beer.name': beer.name, 'beer.hops': beer.hops})
db = DonutStats.instance()
# find our optimal sugar level Donuts above or below this level
# will certainly not be a good match
optimal_sugar_level = db.get_optimal_sugar_level(beer.hops)
return db.get_by_sugar_level(optimal_sugar_level, limit=10)