Skip to content

Instantly share code, notes, and snippets.

@tzaffi
Created June 28, 2020 03:07
Show Gist options
  • Star 1 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save tzaffi/cb3f1b7f238c345a6be462e922c97ba8 to your computer and use it in GitHub Desktop.
Save tzaffi/cb3f1b7f238c345a6be462e922c97ba8 to your computer and use it in GitHub Desktop.
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
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