Created
June 28, 2020 03:07
-
-
Save tzaffi/cb3f1b7f238c345a6be462e922c97ba8 to your computer and use it in GitHub Desktop.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import traceback | |
import requests | |
import sys | |
import linecache | |
import requests | |
# Put your actual Sentry DSN URL here: | |
SENTRY_DSN = 'https://YOUR_SENTRY_KEY@ACCOUNT.ingest.sentry.io/ISSUES_ID' | |
def get_sentry_url_and_header(sentry_dsn, version="7", client_name="my-python/3"): | |
_, _, dsn_url, project_id = sentry_dsn.split('/') | |
key = dsn_url.split('@')[0] | |
url = 'https://sentry.io/api/%s/store/' % project_id | |
auth = 'Sentry sentry_version=%s, sentry_key=%s, sentry_client=%s' % (version, key, client_name) | |
header = { | |
'Content-Type': 'application/json', | |
'X-Sentry-Auth': auth, | |
} | |
return url, header | |
def frame_trans(frame): | |
""" | |
Compatible with both python 2.7 and python 3: | |
""" | |
try: | |
if isinstance(frame, tuple): | |
return { | |
'lineno': frame[1], | |
'context_line': frame[3], | |
'abs_path': frame[0], | |
'function': frame[2], | |
} | |
return { | |
'lineno': frame.lineno, | |
'context_line': frame.line, | |
'abs_path': frame.filename, | |
'function': frame.name, | |
} | |
except: | |
return None | |
def extract_sentry_from_exception(tb): | |
lineno = tb.tb_lineno | |
filename = tb.tb_frame.f_code.co_filename | |
context_line = linecache.getline(filename, lineno, tb.tb_frame.f_globals) | |
func = tb.tb_frame.f_code.co_name | |
return { | |
'lineno': lineno, | |
'context_line': context_line, | |
'abs_path': filename, | |
'function': func, | |
} | |
def extract_all_sentry_frames_from_exception(tb): | |
if tb is None: | |
return [] | |
return [extract_sentry_from_exception(tb)] + extract_all_sentry_frames_from_exception(tb.tb_next) | |
def sentry_friendly_trace(get_last_exception=True): | |
try: | |
current_call = list(map(frame_trans, traceback.extract_stack())) | |
alert_frame = current_call[-4] | |
before_call = current_call[:-4] | |
err_type, err, tb = sys.exc_info() if get_last_exception else (None, None, None) | |
after_call = [alert_frame] if err_type is None else extract_all_sentry_frames_from_exception(tb) | |
return before_call + after_call, err, alert_frame | |
except: | |
return None, None, None | |
def get_sentry_body(message=None, get_last_exception=True): | |
try: | |
trace, exception, alert_frame = sentry_friendly_trace(get_last_exception=get_last_exception) | |
culprit = alert_frame['function'] | |
if message == None: | |
message = "Alert from %s" % culprit | |
exc_json = None | |
if exception is not None: | |
exc_json = { | |
"values": [ | |
{ | |
"type": str(type(exception)), | |
"value": str(exception), | |
"module": str(culprit), | |
} | |
] | |
} | |
return { | |
"culprit": culprit, | |
"message": message, | |
"tags": {"team": "risk"}, | |
"stacktrace": {"frames": trace}, | |
"exception": exc_json, | |
} | |
except: | |
return None | |
def alert_sentry(message=None, get_last_exception=True): | |
url, headers = get_sentry_url_and_header(SENTRY_DSN) | |
json = get_sentry_body(message=message, get_last_exception=get_last_exception) | |
resp = requests.post(url, headers=headers, json=json) | |
return resp |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import post_to_sentry_dsn | |
from time import sleep | |
def _DBZ_inner(): | |
5 / 0 | |
def DBZ_func(): | |
_DBZ_inner() | |
def typical_usage(exception=False, message=None): | |
try: | |
if exception: | |
return DBZ_func() | |
except: | |
return post_to_sentry_dsn.alert_sentry(message=message) | |
return post_to_sentry_dsn.alert_sentry(message=message, get_last_exception=False) | |
def try_alert_examples(N=5): | |
message = "This is a quality assurance test!!! Please Ignore" | |
call = 0 | |
for i in range(N): | |
for exception in [True, False]: | |
for msg in [None, message]: | |
sleep(0.25) | |
call += 1 | |
resp = typical_usage(exception=exception, message=msg) | |
print("%d. (%s, %s) ---> %s" % (call, | |
("" if exception else "no ") + "exception", | |
("" if message is not None else "no ") + "message", | |
resp)) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment