Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified
📁 ..	-	2026-03-27 02:38
📄 __init__.py from sentry_sdk.hub import Hub, init from sentry_sdk.scope import Scope from sentry_sdk.transport import Transport, HttpTransport from sentry_sdk.client import Client from sentry_sdk.api import * # noqa from sentry_sdk.consts import VERSION # noqa from sentry_sdk.crons import monitor # noqa from sentry_sdk.tracing import trace # noqa __all__ = [ # noqa "Hub", "Scope", "Client", "Transport", "HttpTransport", "init", "integrations", # From sentry_sdk.api "capture_event", "capture_message", "capture_exception", "add_breadcrumb", "configure_scope", "push_scope", "flush", "last_event_id", "start_span", "start_transaction", "set_tag", "set_context", "set_extra", "set_user", "set_level", "set_measurement", "get_current_span", "get_traceparent", "get_baggage", "continue_trace", ] # Initialize the debug support after everything is loaded from sentry_sdk.debug import init_debug_support init_debug_support() del init_debug_support	1.03 KB	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 _compat.py import sys from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Optional from typing import Tuple from typing import Any from typing import Type from typing import TypeVar T = TypeVar("T") PY2 = sys.version_info[0] == 2 PY33 = sys.version_info[0] == 3 and sys.version_info[1] >= 3 PY37 = sys.version_info[0] == 3 and sys.version_info[1] >= 7 PY310 = sys.version_info[0] == 3 and sys.version_info[1] >= 10 PY311 = sys.version_info[0] == 3 and sys.version_info[1] >= 11 if PY2: import urlparse text_type = unicode # noqa string_types = (str, text_type) number_types = (int, long, float) # noqa int_types = (int, long) # noqa iteritems = lambda x: x.iteritems() # noqa: B301 binary_sequence_types = (bytearray, memoryview) def implements_str(cls): # type: (T) -> T cls.__unicode__ = cls.__str__ cls.__str__ = lambda x: unicode(x).encode("utf-8") # noqa return cls exec("def reraise(tp, value, tb=None):\n raise tp, value, tb") else: import urllib.parse as urlparse # noqa text_type = str string_types = (text_type,) # type: Tuple[type] number_types = (int, float) # type: Tuple[type, type] int_types = (int,) iteritems = lambda x: x.items() binary_sequence_types = (bytes, bytearray, memoryview) def implements_str(x): # type: (T) -> T return x def reraise(tp, value, tb=None): # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[Any]) -> None assert value is not None if value.__traceback__ is not tb: raise value.with_traceback(tb) raise value def with_metaclass(meta, bases): # type: (Any, Any) -> Any class MetaClass(type): def __new__(metacls, name, this_bases, d): # type: (Any, Any, Any, Any) -> Any return meta(name, bases, d) return type.__new__(MetaClass, "temporary_class", (), {}) def check_thread_support(): # type: () -> None try: from uwsgi import opt # type: ignore except ImportError: return # When `threads` is passed in as a uwsgi option, # `enable-threads` is implied on. if "threads" in opt: return # put here because of circular import from sentry_sdk.consts import FALSE_VALUES if str(opt.get("enable-threads", "0")).lower() in FALSE_VALUES: from warnings import warn warn( Warning( "We detected the use of uwsgi with disabled threads. " "This will cause issues with the transport you are " "trying to use. Please enable threading for uwsgi. " '(Add the "enable-threads" flag).' ) )	2.73 KB	2026-01-20 12:58
📄 _functools.py """ A backport of Python 3 functools to Python 2/3. The only important change we rely upon is that `update_wrapper` handles AttributeError gracefully. Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2 -------------------------------------------- 1. This LICENSE AGREEMENT is between the Python Software Foundation ("PSF"), and the Individual or Organization ("Licensee") accessing and otherwise using this software ("Python") in source or binary form and its associated documentation. 2. Subject to the terms and conditions of this License Agreement, PSF hereby grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce, analyze, test, perform and/or display publicly, prepare derivative works, distribute, and otherwise use Python alone or in any derivative version, provided, however, that PSF's License Agreement and PSF's notice of copyright, i.e., "Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved" are retained in Python alone or in any derivative version prepared by Licensee. 3. In the event Licensee prepares a derivative work that is based on or incorporates Python or any part thereof, and wants to make the derivative work available to others as provided herein, then Licensee hereby agrees to include in any such work a brief summary of the changes made to Python. 4. PSF is making Python available to Licensee on an "AS IS" basis. PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR IMPLIED. BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT INFRINGE ANY THIRD PARTY RIGHTS. 5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON, OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF. 6. This License Agreement will automatically terminate upon a material breach of its terms and conditions. 7. Nothing in this License Agreement shall be deemed to create any relationship of agency, partnership, or joint venture between PSF and Licensee. This License Agreement does not grant permission to use PSF trademarks or trade name in a trademark sense to endorse or promote products or services of Licensee, or any third party. 8. By copying, installing or otherwise using Python, Licensee agrees to be bound by the terms and conditions of this License Agreement. """ from functools import partial from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Any from typing import Callable WRAPPER_ASSIGNMENTS = ( "__module__", "__name__", "__qualname__", "__doc__", "__annotations__", ) WRAPPER_UPDATES = ("__dict__",) def update_wrapper( wrapper, wrapped, assigned=WRAPPER_ASSIGNMENTS, updated=WRAPPER_UPDATES ): # type: (Any, Any, Any, Any) -> Any """Update a wrapper function to look like the wrapped function wrapper is the function to be updated wrapped is the original function assigned is a tuple naming the attributes assigned directly from the wrapped function to the wrapper function (defaults to functools.WRAPPER_ASSIGNMENTS) updated is a tuple naming the attributes of the wrapper that are updated with the corresponding attribute from the wrapped function (defaults to functools.WRAPPER_UPDATES) """ for attr in assigned: try: value = getattr(wrapped, attr) except AttributeError: pass else: setattr(wrapper, attr, value) for attr in updated: getattr(wrapper, attr).update(getattr(wrapped, attr, {})) # Issue #17482: set __wrapped__ last so we don't inadvertently copy it # from the wrapped function when updating __dict__ wrapper.__wrapped__ = wrapped # Return the wrapper so this can be used as a decorator via partial() return wrapper def wraps(wrapped, assigned=WRAPPER_ASSIGNMENTS, updated=WRAPPER_UPDATES): # type: (Callable[..., Any], Any, Any) -> Callable[[Callable[..., Any]], Callable[..., Any]] """Decorator factory to apply update_wrapper() to a wrapper function Returns a decorator that invokes update_wrapper() with the decorated function as the wrapper argument and the arguments to wraps() as the remaining arguments. Default arguments are as for update_wrapper(). This is a convenience function to simplify applying partial() to update_wrapper(). """ return partial(update_wrapper, wrapped=wrapped, assigned=assigned, updated=updated)	4.84 KB	2026-01-20 12:58
📄 _lru_cache.py """ A fork of Python 3.6's stdlib lru_cache (found in Python's 'cpython/Lib/functools.py') adapted into a data structure for single threaded uses. https://github.com/python/cpython/blob/v3.6.12/Lib/functools.py Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2 -------------------------------------------- 1. This LICENSE AGREEMENT is between the Python Software Foundation ("PSF"), and the Individual or Organization ("Licensee") accessing and otherwise using this software ("Python") in source or binary form and its associated documentation. 2. Subject to the terms and conditions of this License Agreement, PSF hereby grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce, analyze, test, perform and/or display publicly, prepare derivative works, distribute, and otherwise use Python alone or in any derivative version, provided, however, that PSF's License Agreement and PSF's notice of copyright, i.e., "Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved" are retained in Python alone or in any derivative version prepared by Licensee. 3. In the event Licensee prepares a derivative work that is based on or incorporates Python or any part thereof, and wants to make the derivative work available to others as provided herein, then Licensee hereby agrees to include in any such work a brief summary of the changes made to Python. 4. PSF is making Python available to Licensee on an "AS IS" basis. PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR IMPLIED. BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT INFRINGE ANY THIRD PARTY RIGHTS. 5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON, OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF. 6. This License Agreement will automatically terminate upon a material breach of its terms and conditions. 7. Nothing in this License Agreement shall be deemed to create any relationship of agency, partnership, or joint venture between PSF and Licensee. This License Agreement does not grant permission to use PSF trademarks or trade name in a trademark sense to endorse or promote products or services of Licensee, or any third party. 8. By copying, installing or otherwise using Python, Licensee agrees to be bound by the terms and conditions of this License Agreement. """ SENTINEL = object() # aliases to the entries in a node PREV = 0 NEXT = 1 KEY = 2 VALUE = 3 class LRUCache(object): def __init__(self, max_size): assert max_size > 0 self.max_size = max_size self.full = False self.cache = {} # root of the circularly linked list to keep track of # the least recently used key self.root = [] # type: ignore # the node looks like [PREV, NEXT, KEY, VALUE] self.root[:] = [self.root, self.root, None, None] self.hits = self.misses = 0 def set(self, key, value): link = self.cache.get(key, SENTINEL) if link is not SENTINEL: # have to move the node to the front of the linked list link_prev, link_next, _key, _value = link # first remove the node from the lsnked list link_prev[NEXT] = link_next link_next[PREV] = link_prev # insert the node between the root and the last last = self.root[PREV] last[NEXT] = self.root[PREV] = link link[PREV] = last link[NEXT] = self.root # update the value link[VALUE] = value elif self.full: # reuse the root node, so update its key/value old_root = self.root old_root[KEY] = key old_root[VALUE] = value self.root = old_root[NEXT] old_key = self.root[KEY] self.root[KEY] = self.root[VALUE] = None del self.cache[old_key] self.cache[key] = old_root else: # insert new node after last last = self.root[PREV] link = [last, self.root, key, value] last[NEXT] = self.root[PREV] = self.cache[key] = link self.full = len(self.cache) >= self.max_size def get(self, key, default=None): link = self.cache.get(key, SENTINEL) if link is SENTINEL: self.misses += 1 return default # have to move the node to the front of the linked list link_prev, link_next, _key, _value = link # first remove the node from the lsnked list link_prev[NEXT] = link_next link_next[PREV] = link_prev # insert the node between the root and the last last = self.root[PREV] last[NEXT] = self.root[PREV] = link link[PREV] = last link[NEXT] = self.root self.hits += 1 return link[VALUE]	5.26 KB	2026-01-20 12:58
📄 _queue.py """ A fork of Python 3.6's stdlib queue (found in Pythons 'cpython/Lib/queue.py') with Lock swapped out for RLock to avoid a deadlock while garbage collecting. https://github.com/python/cpython/blob/v3.6.12/Lib/queue.py See also https://codewithoutrules.com/2017/08/16/concurrency-python/ https://bugs.python.org/issue14976 https://github.com/sqlalchemy/sqlalchemy/blob/4eb747b61f0c1b1c25bdee3856d7195d10a0c227/lib/sqlalchemy/queue.py#L1 We also vendor the code to evade eventlet's broken monkeypatching, see https://github.com/getsentry/sentry-python/pull/484 Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved PYTHON SOFTWARE FOUNDATION LICENSE VERSION 2 -------------------------------------------- 1. This LICENSE AGREEMENT is between the Python Software Foundation ("PSF"), and the Individual or Organization ("Licensee") accessing and otherwise using this software ("Python") in source or binary form and its associated documentation. 2. Subject to the terms and conditions of this License Agreement, PSF hereby grants Licensee a nonexclusive, royalty-free, world-wide license to reproduce, analyze, test, perform and/or display publicly, prepare derivative works, distribute, and otherwise use Python alone or in any derivative version, provided, however, that PSF's License Agreement and PSF's notice of copyright, i.e., "Copyright (c) 2001, 2002, 2003, 2004, 2005, 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013, 2014, 2015, 2016, 2017, 2018, 2019, 2020 Python Software Foundation; All Rights Reserved" are retained in Python alone or in any derivative version prepared by Licensee. 3. In the event Licensee prepares a derivative work that is based on or incorporates Python or any part thereof, and wants to make the derivative work available to others as provided herein, then Licensee hereby agrees to include in any such work a brief summary of the changes made to Python. 4. PSF is making Python available to Licensee on an "AS IS" basis. PSF MAKES NO REPRESENTATIONS OR WARRANTIES, EXPRESS OR IMPLIED. BY WAY OF EXAMPLE, BUT NOT LIMITATION, PSF MAKES NO AND DISCLAIMS ANY REPRESENTATION OR WARRANTY OF MERCHANTABILITY OR FITNESS FOR ANY PARTICULAR PURPOSE OR THAT THE USE OF PYTHON WILL NOT INFRINGE ANY THIRD PARTY RIGHTS. 5. PSF SHALL NOT BE LIABLE TO LICENSEE OR ANY OTHER USERS OF PYTHON FOR ANY INCIDENTAL, SPECIAL, OR CONSEQUENTIAL DAMAGES OR LOSS AS A RESULT OF MODIFYING, DISTRIBUTING, OR OTHERWISE USING PYTHON, OR ANY DERIVATIVE THEREOF, EVEN IF ADVISED OF THE POSSIBILITY THEREOF. 6. This License Agreement will automatically terminate upon a material breach of its terms and conditions. 7. Nothing in this License Agreement shall be deemed to create any relationship of agency, partnership, or joint venture between PSF and Licensee. This License Agreement does not grant permission to use PSF trademarks or trade name in a trademark sense to endorse or promote products or services of Licensee, or any third party. 8. By copying, installing or otherwise using Python, Licensee agrees to be bound by the terms and conditions of this License Agreement. """ import threading from collections import deque from time import time from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Any __all__ = ["EmptyError", "FullError", "Queue"] class EmptyError(Exception): "Exception raised by Queue.get(block=0)/get_nowait()." pass class FullError(Exception): "Exception raised by Queue.put(block=0)/put_nowait()." pass class Queue(object): """Create a queue object with a given maximum size. If maxsize is <= 0, the queue size is infinite. """ def __init__(self, maxsize=0): self.maxsize = maxsize self._init(maxsize) # mutex must be held whenever the queue is mutating. All methods # that acquire mutex must release it before returning. mutex # is shared between the three conditions, so acquiring and # releasing the conditions also acquires and releases mutex. self.mutex = threading.RLock() # Notify not_empty whenever an item is added to the queue; a # thread waiting to get is notified then. self.not_empty = threading.Condition(self.mutex) # Notify not_full whenever an item is removed from the queue; # a thread waiting to put is notified then. self.not_full = threading.Condition(self.mutex) # Notify all_tasks_done whenever the number of unfinished tasks # drops to zero; thread waiting to join() is notified to resume self.all_tasks_done = threading.Condition(self.mutex) self.unfinished_tasks = 0 def task_done(self): """Indicate that a formerly enqueued task is complete. Used by Queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete. If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue). Raises a ValueError if called more times than there were items placed in the queue. """ with self.all_tasks_done: unfinished = self.unfinished_tasks - 1 if unfinished <= 0: if unfinished < 0: raise ValueError("task_done() called too many times") self.all_tasks_done.notify_all() self.unfinished_tasks = unfinished def join(self): """Blocks until all items in the Queue have been gotten and processed. The count of unfinished tasks goes up whenever an item is added to the queue. The count goes down whenever a consumer thread calls task_done() to indicate the item was retrieved and all work on it is complete. When the count of unfinished tasks drops to zero, join() unblocks. """ with self.all_tasks_done: while self.unfinished_tasks: self.all_tasks_done.wait() def qsize(self): """Return the approximate size of the queue (not reliable!).""" with self.mutex: return self._qsize() def empty(self): """Return True if the queue is empty, False otherwise (not reliable!). This method is likely to be removed at some point. Use qsize() == 0 as a direct substitute, but be aware that either approach risks a race condition where a queue can grow before the result of empty() or qsize() can be used. To create code that needs to wait for all queued tasks to be completed, the preferred technique is to use the join() method. """ with self.mutex: return not self._qsize() def full(self): """Return True if the queue is full, False otherwise (not reliable!). This method is likely to be removed at some point. Use qsize() >= n as a direct substitute, but be aware that either approach risks a race condition where a queue can shrink before the result of full() or qsize() can be used. """ with self.mutex: return 0 < self.maxsize <= self._qsize() def put(self, item, block=True, timeout=None): """Put an item into the queue. If optional args 'block' is true and 'timeout' is None (the default), block if necessary until a free slot is available. If 'timeout' is a non-negative number, it blocks at most 'timeout' seconds and raises the FullError exception if no free slot was available within that time. Otherwise ('block' is false), put an item on the queue if a free slot is immediately available, else raise the FullError exception ('timeout' is ignored in that case). """ with self.not_full: if self.maxsize > 0: if not block: if self._qsize() >= self.maxsize: raise FullError() elif timeout is None: while self._qsize() >= self.maxsize: self.not_full.wait() elif timeout < 0: raise ValueError("'timeout' must be a non-negative number") else: endtime = time() + timeout while self._qsize() >= self.maxsize: remaining = endtime - time() if remaining <= 0.0: raise FullError() self.not_full.wait(remaining) self._put(item) self.unfinished_tasks += 1 self.not_empty.notify() def get(self, block=True, timeout=None): """Remove and return an item from the queue. If optional args 'block' is true and 'timeout' is None (the default), block if necessary until an item is available. If 'timeout' is a non-negative number, it blocks at most 'timeout' seconds and raises the EmptyError exception if no item was available within that time. Otherwise ('block' is false), return an item if one is immediately available, else raise the EmptyError exception ('timeout' is ignored in that case). """ with self.not_empty: if not block: if not self._qsize(): raise EmptyError() elif timeout is None: while not self._qsize(): self.not_empty.wait() elif timeout < 0: raise ValueError("'timeout' must be a non-negative number") else: endtime = time() + timeout while not self._qsize(): remaining = endtime - time() if remaining <= 0.0: raise EmptyError() self.not_empty.wait(remaining) item = self._get() self.not_full.notify() return item def put_nowait(self, item): """Put an item into the queue without blocking. Only enqueue the item if a free slot is immediately available. Otherwise raise the FullError exception. """ return self.put(item, block=False) def get_nowait(self): """Remove and return an item from the queue without blocking. Only get an item if one is immediately available. Otherwise raise the EmptyError exception. """ return self.get(block=False) # Override these methods to implement other queue organizations # (e.g. stack or priority queue). # These will only be called with appropriate locks held # Initialize the queue representation def _init(self, maxsize): self.queue = deque() # type: Any def _qsize(self): return len(self.queue) # Put a new item in the queue def _put(self, item): self.queue.append(item) # Get an item from the queue def _get(self): return self.queue.popleft()	11 KB	2026-01-20 12:58
📄 _types.py try: from typing import TYPE_CHECKING as TYPE_CHECKING except ImportError: TYPE_CHECKING = False # Re-exported for compat, since code out there in the wild might use this variable. MYPY = TYPE_CHECKING if TYPE_CHECKING: from types import TracebackType from typing import Any from typing import Callable from typing import Dict from typing import Optional from typing import Tuple from typing import Type from typing import Union from typing_extensions import Literal ExcInfo = Tuple[ Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType] ] Event = Dict[str, Any] Hint = Dict[str, Any] Breadcrumb = Dict[str, Any] BreadcrumbHint = Dict[str, Any] SamplingContext = Dict[str, Any] EventProcessor = Callable[[Event, Hint], Optional[Event]] ErrorProcessor = Callable[[Event, ExcInfo], Optional[Event]] BreadcrumbProcessor = Callable[[Breadcrumb, BreadcrumbHint], Optional[Breadcrumb]] TransactionProcessor = Callable[[Event, Hint], Optional[Event]] TracesSampler = Callable[[SamplingContext], Union[float, int, bool]] # https://github.com/python/mypy/issues/5710 NotImplementedType = Any EventDataCategory = Literal[ "default", "error", "crash", "transaction", "security", "attachment", "session", "internal", "profile", ] SessionStatus = Literal["ok", "exited", "crashed", "abnormal"] EndpointType = Literal["store", "envelope"] DurationUnit = Literal[ "nanosecond", "microsecond", "millisecond", "second", "minute", "hour", "day", "week", ] InformationUnit = Literal[ "bit", "byte", "kilobyte", "kibibyte", "megabyte", "mebibyte", "gigabyte", "gibibyte", "terabyte", "tebibyte", "petabyte", "pebibyte", "exabyte", "exbibyte", ] FractionUnit = Literal["ratio", "percent"] MeasurementUnit = Union[DurationUnit, InformationUnit, FractionUnit, str] ProfilerMode = Literal["sleep", "thread", "gevent", "unknown"]	2.19 KB	2026-01-20 12:58
📄 _werkzeug.py """ Copyright (c) 2007 by the Pallets team. Some rights reserved. Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. * Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. THIS SOFTWARE AND DOCUMENTATION IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE AND DOCUMENTATION, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """ from sentry_sdk._compat import iteritems from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Dict from typing import Iterator from typing import Tuple # # `get_headers` comes from `werkzeug.datastructures.EnvironHeaders` # https://github.com/pallets/werkzeug/blob/0.14.1/werkzeug/datastructures.py#L1361 # # We need this function because Django does not give us a "pure" http header # dict. So we might as well use it for all WSGI integrations. # def _get_headers(environ): # type: (Dict[str, str]) -> Iterator[Tuple[str, str]] """ Returns only proper HTTP headers. """ for key, value in iteritems(environ): key = str(key) if key.startswith("HTTP_") and key not in ( "HTTP_CONTENT_TYPE", "HTTP_CONTENT_LENGTH", ): yield key[5:].replace("_", "-").title(), value elif key in ("CONTENT_TYPE", "CONTENT_LENGTH"): yield key.replace("_", "-").title(), value # # `get_host` comes from `werkzeug.wsgi.get_host` # https://github.com/pallets/werkzeug/blob/1.0.1/src/werkzeug/wsgi.py#L145 # def get_host(environ, use_x_forwarded_for=False): # type: (Dict[str, str], bool) -> str """ Return the host for the given WSGI environment. """ if use_x_forwarded_for and "HTTP_X_FORWARDED_HOST" in environ: rv = environ["HTTP_X_FORWARDED_HOST"] if environ["wsgi.url_scheme"] == "http" and rv.endswith(":80"): rv = rv[:-3] elif environ["wsgi.url_scheme"] == "https" and rv.endswith(":443"): rv = rv[:-4] elif environ.get("HTTP_HOST"): rv = environ["HTTP_HOST"] if environ["wsgi.url_scheme"] == "http" and rv.endswith(":80"): rv = rv[:-3] elif environ["wsgi.url_scheme"] == "https" and rv.endswith(":443"): rv = rv[:-4] elif environ.get("SERVER_NAME"): rv = environ["SERVER_NAME"] if (environ["wsgi.url_scheme"], environ["SERVER_PORT"]) not in ( ("https", "443"), ("http", "80"), ): rv += ":" + environ["SERVER_PORT"] else: # In spite of the WSGI spec, SERVER_NAME might not be present. rv = "unknown" return rv	3.7 KB	2026-01-20 12:58
📄 api.py import inspect from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.hub import Hub from sentry_sdk.scope import Scope from sentry_sdk.tracing import NoOpSpan, Transaction if TYPE_CHECKING: from typing import Any from typing import Dict from typing import Optional from typing import overload from typing import Callable from typing import TypeVar from typing import ContextManager from typing import Union from sentry_sdk._types import ( Event, Hint, Breadcrumb, BreadcrumbHint, ExcInfo, MeasurementUnit, ) from sentry_sdk.tracing import Span T = TypeVar("T") F = TypeVar("F", bound=Callable[..., Any]) else: def overload(x): # type: (T) -> T return x # When changing this, update __all__ in __init__.py too __all__ = [ "capture_event", "capture_message", "capture_exception", "add_breadcrumb", "configure_scope", "push_scope", "flush", "last_event_id", "start_span", "start_transaction", "set_tag", "set_context", "set_extra", "set_user", "set_level", "set_measurement", "get_current_span", "get_traceparent", "get_baggage", "continue_trace", ] def hubmethod(f): # type: (F) -> F f.__doc__ = "%s\n\n%s" % ( "Alias for :py:meth:`sentry_sdk.Hub.%s`" % f.__name__, inspect.getdoc(getattr(Hub, f.__name__)), ) return f def scopemethod(f): # type: (F) -> F f.__doc__ = "%s\n\n%s" % ( "Alias for :py:meth:`sentry_sdk.Scope.%s`" % f.__name__, inspect.getdoc(getattr(Scope, f.__name__)), ) return f @hubmethod def capture_event( event, # type: Event hint=None, # type: Optional[Hint] scope=None, # type: Optional[Any] scope_args # type: Any ): # type: (...) -> Optional[str] return Hub.current.capture_event(event, hint, scope=scope, scope_args) @hubmethod def capture_message( message, # type: str level=None, # type: Optional[str] scope=None, # type: Optional[Any] scope_args # type: Any ): # type: (...) -> Optional[str] return Hub.current.capture_message(message, level, scope=scope, scope_args) @hubmethod def capture_exception( error=None, # type: Optional[Union[BaseException, ExcInfo]] scope=None, # type: Optional[Any] scope_args # type: Any ): # type: (...) -> Optional[str] return Hub.current.capture_exception(error, scope=scope, scope_args) @hubmethod def add_breadcrumb( crumb=None, # type: Optional[Breadcrumb] hint=None, # type: Optional[BreadcrumbHint] kwargs # type: Any ): # type: (...) -> None return Hub.current.add_breadcrumb(crumb, hint, kwargs) @overload def configure_scope(): # type: () -> ContextManager[Scope] pass @overload def configure_scope( # noqa: F811 callback, # type: Callable[[Scope], None] ): # type: (...) -> None pass @hubmethod def configure_scope( # noqa: F811 callback=None, # type: Optional[Callable[[Scope], None]] ): # type: (...) -> Optional[ContextManager[Scope]] return Hub.current.configure_scope(callback) @overload def push_scope(): # type: () -> ContextManager[Scope] pass @overload def push_scope( # noqa: F811 callback, # type: Callable[[Scope], None] ): # type: (...) -> None pass @hubmethod def push_scope( # noqa: F811 callback=None, # type: Optional[Callable[[Scope], None]] ): # type: (...) -> Optional[ContextManager[Scope]] return Hub.current.push_scope(callback) @scopemethod def set_tag(key, value): # type: (str, Any) -> None return Hub.current.scope.set_tag(key, value) @scopemethod def set_context(key, value): # type: (str, Dict[str, Any]) -> None return Hub.current.scope.set_context(key, value) @scopemethod def set_extra(key, value): # type: (str, Any) -> None return Hub.current.scope.set_extra(key, value) @scopemethod def set_user(value): # type: (Optional[Dict[str, Any]]) -> None return Hub.current.scope.set_user(value) @scopemethod def set_level(value): # type: (str) -> None return Hub.current.scope.set_level(value) @hubmethod def flush( timeout=None, # type: Optional[float] callback=None, # type: Optional[Callable[[int, float], None]] ): # type: (...) -> None return Hub.current.flush(timeout=timeout, callback=callback) @hubmethod def last_event_id(): # type: () -> Optional[str] return Hub.current.last_event_id() @hubmethod def start_span( span=None, # type: Optional[Span] kwargs # type: Any ): # type: (...) -> Span return Hub.current.start_span(span=span, kwargs) @hubmethod def start_transaction( transaction=None, # type: Optional[Transaction] kwargs # type: Any ): # type: (...) -> Union[Transaction, NoOpSpan] return Hub.current.start_transaction(transaction, kwargs) def set_measurement(name, value, unit=""): # type: (str, float, MeasurementUnit) -> None transaction = Hub.current.scope.transaction if transaction is not None: transaction.set_measurement(name, value, unit) def get_current_span(hub=None): # type: (Optional[Hub]) -> Optional[Span] """ Returns the currently active span if there is one running, otherwise `None` """ if hub is None: hub = Hub.current current_span = hub.scope.span return current_span def get_traceparent(): # type: () -> Optional[str] """ Returns the traceparent either from the active span or from the scope. """ return Hub.current.get_traceparent() def get_baggage(): # type: () -> Optional[str] """ Returns Baggage either from the active span or from the scope. """ return Hub.current.get_baggage() def continue_trace(environ_or_headers, op=None, name=None, source=None): # type: (Dict[str, Any], Optional[str], Optional[str], Optional[str]) -> Transaction """ Sets the propagation context from environment or headers and returns a transaction. """ return Hub.current.continue_trace(environ_or_headers, op, name, source)	6.05 KB	2026-01-20 12:58
📄 attachments.py import os import mimetypes from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.envelope import Item, PayloadRef if TYPE_CHECKING: from typing import Optional, Union, Callable class Attachment(object): def __init__( self, bytes=None, # type: Union[None, bytes, Callable[[], bytes]] filename=None, # type: Optional[str] path=None, # type: Optional[str] content_type=None, # type: Optional[str] add_to_transactions=False, # type: bool ): # type: (...) -> None if bytes is None and path is None: raise TypeError("path or raw bytes required for attachment") if filename is None and path is not None: filename = os.path.basename(path) if filename is None: raise TypeError("filename is required for attachment") if content_type is None: content_type = mimetypes.guess_type(filename)[0] self.bytes = bytes self.filename = filename self.path = path self.content_type = content_type self.add_to_transactions = add_to_transactions def to_envelope_item(self): # type: () -> Item """Returns an envelope item for this attachment.""" payload = None # type: Union[None, PayloadRef, bytes] if self.bytes is not None: if callable(self.bytes): payload = self.bytes() else: payload = self.bytes else: payload = PayloadRef(path=self.path) return Item( payload=payload, type="attachment", content_type=self.content_type, filename=self.filename, ) def __repr__(self): # type: () -> str return "<Attachment %r>" % (self.filename,)	1.77 KB	2026-01-20 12:58
📄 client.py from importlib import import_module import os import uuid import random from datetime import datetime import socket from sentry_sdk._compat import string_types, text_type, iteritems from sentry_sdk.utils import ( capture_internal_exceptions, current_stacktrace, disable_capture_event, format_timestamp, get_sdk_name, get_type_name, get_default_release, handle_in_app, logger, ) from sentry_sdk.serializer import serialize from sentry_sdk.tracing import trace, has_tracing_enabled from sentry_sdk.transport import make_transport from sentry_sdk.consts import ( DEFAULT_MAX_VALUE_LENGTH, DEFAULT_OPTIONS, INSTRUMENTER, VERSION, ClientConstructor, ) from sentry_sdk.integrations import setup_integrations from sentry_sdk.utils import ContextVar from sentry_sdk.sessions import SessionFlusher from sentry_sdk.envelope import Envelope from sentry_sdk.profiler import has_profiling_enabled, setup_profiler from sentry_sdk.scrubber import EventScrubber from sentry_sdk.monitor import Monitor from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Any from typing import Callable from typing import Dict from typing import Optional from typing import Sequence from sentry_sdk.scope import Scope from sentry_sdk._types import Event, Hint from sentry_sdk.session import Session _client_init_debug = ContextVar("client_init_debug") SDK_INFO = { "name": "sentry.python", # SDK name will be overridden after integrations have been loaded with sentry_sdk.integrations.setup_integrations() "version": VERSION, "packages": [{"name": "pypi:sentry-sdk", "version": VERSION}], } def _get_options(args, kwargs): # type: (Optional[str], *Any) -> Dict[str, Any] if args and (isinstance(args[0], (text_type, bytes, str)) or args[0] is None): dsn = args[0] # type: Optional[str] args = args[1:] else: dsn = None if len(args) > 1: raise TypeError("Only single positional argument is expected") rv = dict(DEFAULT_OPTIONS) options = dict(args, *kwargs) if dsn is not None and options.get("dsn") is None: options["dsn"] = dsn for key, value in iteritems(options): if key not in rv: # Option "with_locals" was renamed to "include_local_variables" if key == "with_locals": msg = ( "Deprecated: The option 'with_locals' was renamed to 'include_local_variables'. " "Please use 'include_local_variables'. The option 'with_locals' will be removed in the future." ) logger.warning(msg) rv["include_local_variables"] = value continue # Option "request_bodies" was renamed to "max_request_body_size" if key == "request_bodies": msg = ( "Deprecated: The option 'request_bodies' was renamed to 'max_request_body_size'. " "Please use 'max_request_body_size'. The option 'request_bodies' will be removed in the future." ) logger.warning(msg) rv["max_request_body_size"] = value continue raise TypeError("Unknown option %r" % (key,)) rv[key] = value if rv["dsn"] is None: rv["dsn"] = os.environ.get("SENTRY_DSN") if rv["release"] is None: rv["release"] = get_default_release() if rv["environment"] is None: rv["environment"] = os.environ.get("SENTRY_ENVIRONMENT") or "production" if rv["server_name"] is None and hasattr(socket, "gethostname"): rv["server_name"] = socket.gethostname() if rv["instrumenter"] is None: rv["instrumenter"] = INSTRUMENTER.SENTRY if rv["project_root"] is None: try: project_root = os.getcwd() except Exception: project_root = None rv["project_root"] = project_root if rv["enable_tracing"] is True and rv["traces_sample_rate"] is None: rv["traces_sample_rate"] = 1.0 if rv["event_scrubber"] is None: rv["event_scrubber"] = EventScrubber() return rv try: # Python 3.6+ module_not_found_error = ModuleNotFoundError except Exception: # Older Python versions module_not_found_error = ImportError # type: ignore class _Client(object): """The client is internally responsible for capturing the events and forwarding them to sentry through the configured transport. It takes the client options as keyword arguments and optionally the DSN as first argument. """ def __init__(self, args, *kwargs): # type: (Any, *Any) -> None self.options = get_options(args, **kwargs) # type: Dict[str, Any] self._init_impl() def __getstate__(self): # type: () -> Any return {"options": self.options} def __setstate__(self, state): # type: (Any) -> None self.options = state["options"] self._init_impl() def _setup_instrumentation(self, functions_to_trace): # type: (Sequence[Dict[str, str]]) -> None """ Instruments the functions given in the list `functions_to_trace` with the `@sentry_sdk.tracing.trace` decorator. """ for function in functions_to_trace: class_name = None function_qualname = function["qualified_name"] module_name, function_name = function_qualname.rsplit(".", 1) try: # Try to import module and function # ex: "mymodule.submodule.funcname" module_obj = import_module(module_name) function_obj = getattr(module_obj, function_name) setattr(module_obj, function_name, trace(function_obj)) logger.debug("Enabled tracing for %s", function_qualname) except module_not_found_error: try: # Try to import a class # ex: "mymodule.submodule.MyClassName.member_function" module_name, class_name = module_name.rsplit(".", 1) module_obj = import_module(module_name) class_obj = getattr(module_obj, class_name) function_obj = getattr(class_obj, function_name) setattr(class_obj, function_name, trace(function_obj)) setattr(module_obj, class_name, class_obj) logger.debug("Enabled tracing for %s", function_qualname) except Exception as e: logger.warning( "Can not enable tracing for '%s'. (%s) Please check your `functions_to_trace` parameter.", function_qualname, e, ) except Exception as e: logger.warning( "Can not enable tracing for '%s'. (%s) Please check your `functions_to_trace` parameter.", function_qualname, e, ) def _init_impl(self): # type: () -> None old_debug = _client_init_debug.get(False) def _capture_envelope(envelope): # type: (Envelope) -> None if self.transport is not None: self.transport.capture_envelope(envelope) try: _client_init_debug.set(self.options["debug"]) self.transport = make_transport(self.options) self.monitor = None if self.transport: if self.options["_experiments"].get( "enable_backpressure_handling", False ): self.monitor = Monitor(self.transport) self.session_flusher = SessionFlusher(capture_func=_capture_envelope) max_request_body_size = ("always", "never", "small", "medium") if self.options["max_request_body_size"] not in max_request_body_size: raise ValueError( "Invalid value for max_request_body_size. Must be one of {}".format( max_request_body_size ) ) self.integrations = setup_integrations( self.options["integrations"], with_defaults=self.options["default_integrations"], with_auto_enabling_integrations=self.options[ "auto_enabling_integrations" ], ) sdk_name = get_sdk_name(list(self.integrations.keys())) SDK_INFO["name"] = sdk_name logger.debug("Setting SDK name to '%s'", sdk_name) finally: _client_init_debug.set(old_debug) if has_profiling_enabled(self.options): try: setup_profiler(self.options) except ValueError as e: logger.debug(str(e)) self._setup_instrumentation(self.options.get("functions_to_trace", [])) @property def dsn(self): # type: () -> Optional[str] """Returns the configured DSN as string.""" return self.options["dsn"] def _prepare_event( self, event, # type: Event hint, # type: Hint scope, # type: Optional[Scope] ): # type: (...) -> Optional[Event] if event.get("timestamp") is None: event["timestamp"] = datetime.utcnow() if scope is not None: is_transaction = event.get("type") == "transaction" event_ = scope.apply_to_event(event, hint, self.options) # one of the event/error processors returned None if event_ is None: if self.transport: self.transport.record_lost_event( "event_processor", data_category=("transaction" if is_transaction else "error"), ) return None event = event_ if ( self.options["attach_stacktrace"] and "exception" not in event and "stacktrace" not in event and "threads" not in event ): with capture_internal_exceptions(): event["threads"] = { "values": [ { "stacktrace": current_stacktrace( include_local_variables=self.options.get( "include_local_variables", True ), max_value_length=self.options.get( "max_value_length", DEFAULT_MAX_VALUE_LENGTH ), ), "crashed": False, "current": True, } ] } for key in "release", "environment", "server_name", "dist": if event.get(key) is None and self.options[key] is not None: event[key] = text_type(self.options[key]).strip() if event.get("sdk") is None: sdk_info = dict(SDK_INFO) sdk_info["integrations"] = sorted(self.integrations.keys()) event["sdk"] = sdk_info if event.get("platform") is None: event["platform"] = "python" event = handle_in_app( event, self.options["in_app_exclude"], self.options["in_app_include"], self.options["project_root"], ) if event is not None: event_scrubber = self.options["event_scrubber"] if event_scrubber and not self.options["send_default_pii"]: event_scrubber.scrub_event(event) # Postprocess the event here so that annotated types do # generally not surface in before_send if event is not None: event = serialize( event, max_request_body_size=self.options.get("max_request_body_size"), max_value_length=self.options.get("max_value_length"), ) before_send = self.options["before_send"] if ( before_send is not None and event is not None and event.get("type") != "transaction" ): new_event = None with capture_internal_exceptions(): new_event = before_send(event, hint or {}) if new_event is None: logger.info("before send dropped event") if self.transport: self.transport.record_lost_event( "before_send", data_category="error" ) event = new_event # type: ignore before_send_transaction = self.options["before_send_transaction"] if ( before_send_transaction is not None and event is not None and event.get("type") == "transaction" ): new_event = None with capture_internal_exceptions(): new_event = before_send_transaction(event, hint or {}) if new_event is None: logger.info("before send transaction dropped event") if self.transport: self.transport.record_lost_event( "before_send", data_category="transaction" ) event = new_event # type: ignore return event def _is_ignored_error(self, event, hint): # type: (Event, Hint) -> bool exc_info = hint.get("exc_info") if exc_info is None: return False error = exc_info[0] error_type_name = get_type_name(exc_info[0]) error_full_name = "%s.%s" % (exc_info[0].__module__, error_type_name) for ignored_error in self.options["ignore_errors"]: # String types are matched against the type name in the # exception only if isinstance(ignored_error, string_types): if ignored_error == error_full_name or ignored_error == error_type_name: return True else: if issubclass(error, ignored_error): return True return False def _should_capture( self, event, # type: Event hint, # type: Hint scope=None, # type: Optional[Scope] ): # type: (...) -> bool # Transactions are sampled independent of error events. is_transaction = event.get("type") == "transaction" if is_transaction: return True ignoring_prevents_recursion = scope is not None and not scope._should_capture if ignoring_prevents_recursion: return False ignored_by_config_option = self._is_ignored_error(event, hint) if ignored_by_config_option: return False return True def _should_sample_error( self, event, # type: Event ): # type: (...) -> bool not_in_sample_rate = ( self.options["sample_rate"] < 1.0 and random.random() >= self.options["sample_rate"] ) if not_in_sample_rate: # because we will not sample this event, record a "lost event". if self.transport: self.transport.record_lost_event("sample_rate", data_category="error") return False return True def _update_session_from_event( self, session, # type: Session event, # type: Event ): # type: (...) -> None crashed = False errored = False user_agent = None exceptions = (event.get("exception") or {}).get("values") if exceptions: errored = True for error in exceptions: mechanism = error.get("mechanism") if mechanism and mechanism.get("handled") is False: crashed = True break user = event.get("user") if session.user_agent is None: headers = (event.get("request") or {}).get("headers") for k, v in iteritems(headers or {}): if k.lower() == "user-agent": user_agent = v break session.update( status="crashed" if crashed else None, user=user, user_agent=user_agent, errors=session.errors + (errored or crashed), ) def capture_event( self, event, # type: Event hint=None, # type: Optional[Hint] scope=None, # type: Optional[Scope] ): # type: (...) -> Optional[str] """Captures an event. :param event: A ready-made event that can be directly sent to Sentry. :param hint: Contains metadata about the event that can be read from `before_send`, such as the original exception object or a HTTP request object. :param scope: An optional scope to use for determining whether this event should be captured. :returns: An event ID. May be `None` if there is no DSN set or of if the SDK decided to discard the event for other reasons. In such situations setting `debug=True` on `init()` may help. """ if disable_capture_event.get(False): return None if self.transport is None: return None if hint is None: hint = {} event_id = event.get("event_id") hint = dict(hint or ()) # type: Hint if event_id is None: event["event_id"] = event_id = uuid.uuid4().hex if not self._should_capture(event, hint, scope): return None profile = event.pop("profile", None) event_opt = self._prepare_event(event, hint, scope) if event_opt is None: return None # whenever we capture an event we also check if the session needs # to be updated based on that information. session = scope._session if scope else None if session: self._update_session_from_event(session, event) is_transaction = event_opt.get("type") == "transaction" is_checkin = event_opt.get("type") == "check_in" if ( not is_transaction and not is_checkin and not self._should_sample_error(event) ): return None tracing_enabled = has_tracing_enabled(self.options) attachments = hint.get("attachments") trace_context = event_opt.get("contexts", {}).get("trace") or {} dynamic_sampling_context = trace_context.pop("dynamic_sampling_context", {}) # If tracing is enabled all events should go to /envelope endpoint. # If no tracing is enabled only transactions, events with attachments, and checkins should go to the /envelope endpoint. should_use_envelope_endpoint = ( tracing_enabled or is_transaction or is_checkin or bool(attachments) ) if should_use_envelope_endpoint: headers = { "event_id": event_opt["event_id"], "sent_at": format_timestamp(datetime.utcnow()), } if dynamic_sampling_context: headers["trace"] = dynamic_sampling_context envelope = Envelope(headers=headers) if is_transaction: if profile is not None: envelope.add_profile(profile.to_json(event_opt, self.options)) envelope.add_transaction(event_opt) elif is_checkin: envelope.add_checkin(event_opt) else: envelope.add_event(event_opt) for attachment in attachments or (): envelope.add_item(attachment.to_envelope_item()) self.transport.capture_envelope(envelope) else: # All other events go to the legacy /store/ endpoint (will be removed in the future). self.transport.capture_event(event_opt) return event_id def capture_session( self, session # type: Session ): # type: (...) -> None if not session.release: logger.info("Discarded session update because of missing release") else: self.session_flusher.add_session(session) def close( self, timeout=None, # type: Optional[float] callback=None, # type: Optional[Callable[[int, float], None]] ): # type: (...) -> None """ Close the client and shut down the transport. Arguments have the same semantics as :py:meth:`Client.flush`. """ if self.transport is not None: self.flush(timeout=timeout, callback=callback) self.session_flusher.kill() if self.monitor: self.monitor.kill() self.transport.kill() self.transport = None def flush( self, timeout=None, # type: Optional[float] callback=None, # type: Optional[Callable[[int, float], None]] ): # type: (...) -> None """ Wait for the current events to be sent. :param timeout: Wait for at most `timeout` seconds. If no `timeout` is provided, the `shutdown_timeout` option value is used. :param callback: Is invoked with the number of pending events and the configured timeout. """ if self.transport is not None: if timeout is None: timeout = self.options["shutdown_timeout"] self.session_flusher.flush() self.transport.flush(timeout=timeout, callback=callback) def __enter__(self): # type: () -> _Client return self def __exit__(self, exc_type, exc_value, tb): # type: (Any, Any, Any) -> None self.close() from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: # Make mypy, PyCharm and other static analyzers think `get_options` is a # type to have nicer autocompletion for params. # # Use `ClientConstructor` to define the argument types of `init` and # `Dict[str, Any]` to tell static analyzers about the return type. class get_options(ClientConstructor, Dict[str, Any]): # noqa: N801 pass class Client(ClientConstructor, _Client): pass else: # Alias `get_options` for actual usage. Go through the lambda indirection # to throw PyCharm off of the weakly typed signature (it would otherwise # discover both the weakly typed signature of `_init` and our faked `init` # type). get_options = (lambda: _get_options)() Client = (lambda: _Client)()	22.35 KB	2026-01-20 12:58
📄 consts.py from sentry_sdk._types import TYPE_CHECKING # up top to prevent circular import due to integration import DEFAULT_MAX_VALUE_LENGTH = 1024 if TYPE_CHECKING: import sentry_sdk from typing import Optional from typing import Callable from typing import Union from typing import List from typing import Type from typing import Dict from typing import Any from typing import Sequence from typing_extensions import TypedDict from sentry_sdk.integrations import Integration from sentry_sdk._types import ( BreadcrumbProcessor, Event, EventProcessor, ProfilerMode, TracesSampler, TransactionProcessor, ) # Experiments are feature flags to enable and disable certain unstable SDK # functionality. Changing them from the defaults (`None`) in production # code is highly discouraged. They are not subject to any stability # guarantees such as the ones from semantic versioning. Experiments = TypedDict( "Experiments", { "max_spans": Optional[int], "record_sql_params": Optional[bool], # TODO: Remove these 2 profiling related experiments "profiles_sample_rate": Optional[float], "profiler_mode": Optional[ProfilerMode], "enable_backpressure_handling": Optional[bool], }, total=False, ) DEFAULT_QUEUE_SIZE = 100 DEFAULT_MAX_BREADCRUMBS = 100 MATCH_ALL = r"." FALSE_VALUES = [ "false", "no", "off", "n", "0", ] class INSTRUMENTER: SENTRY = "sentry" OTEL = "otel" class SPANDATA: """ Additional information describing the type of the span. See: https://develop.sentry.dev/sdk/performance/span-data-conventions/ """ DB_NAME = "db.name" """ The name of the database being accessed. For commands that switch the database, this should be set to the target database (even if the command fails). Example: myDatabase """ DB_OPERATION = "db.operation" """ The name of the operation being executed, e.g. the MongoDB command name such as findAndModify, or the SQL keyword. See: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/database.md Example: findAndModify, HMSET, SELECT """ DB_SYSTEM = "db.system" """ An identifier for the database management system (DBMS) product being used. See: https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/semantic_conventions/database.md Example: postgresql """ CACHE_HIT = "cache.hit" """ A boolean indicating whether the requested data was found in the cache. Example: true """ CACHE_ITEM_SIZE = "cache.item_size" """ The size of the requested data in bytes. Example: 58 """ HTTP_QUERY = "http.query" """ The Query string present in the URL. Example: ?foo=bar&bar=baz """ HTTP_FRAGMENT = "http.fragment" """ The Fragments present in the URL. Example: #foo=bar """ HTTP_METHOD = "http.method" """ The HTTP method used. Example: GET """ HTTP_STATUS_CODE = "http.response.status_code" """ The HTTP status code as an integer. Example: 418 """ SERVER_ADDRESS = "server.address" """ Name of the database host. Example: example.com """ SERVER_PORT = "server.port" """ Logical server port number Example: 80; 8080; 443 """ SERVER_SOCKET_ADDRESS = "server.socket.address" """ Physical server IP address or Unix socket address. Example: 10.5.3.2 """ SERVER_SOCKET_PORT = "server.socket.port" """ Physical server port. Recommended: If different than server.port. Example: 16456 """ class OP: CACHE_GET_ITEM = "cache.get_item" DB = "db" DB_REDIS = "db.redis" EVENT_DJANGO = "event.django" FUNCTION = "function" FUNCTION_AWS = "function.aws" FUNCTION_GCP = "function.gcp" GRPC_CLIENT = "grpc.client" GRPC_SERVER = "grpc.server" HTTP_CLIENT = "http.client" HTTP_CLIENT_STREAM = "http.client.stream" HTTP_SERVER = "http.server" MIDDLEWARE_DJANGO = "middleware.django" MIDDLEWARE_STARLETTE = "middleware.starlette" MIDDLEWARE_STARLETTE_RECEIVE = "middleware.starlette.receive" MIDDLEWARE_STARLETTE_SEND = "middleware.starlette.send" MIDDLEWARE_STARLITE = "middleware.starlite" MIDDLEWARE_STARLITE_RECEIVE = "middleware.starlite.receive" MIDDLEWARE_STARLITE_SEND = "middleware.starlite.send" QUEUE_SUBMIT_ARQ = "queue.submit.arq" QUEUE_TASK_ARQ = "queue.task.arq" QUEUE_SUBMIT_CELERY = "queue.submit.celery" QUEUE_TASK_CELERY = "queue.task.celery" QUEUE_TASK_RQ = "queue.task.rq" QUEUE_SUBMIT_HUEY = "queue.submit.huey" QUEUE_TASK_HUEY = "queue.task.huey" SUBPROCESS = "subprocess" SUBPROCESS_WAIT = "subprocess.wait" SUBPROCESS_COMMUNICATE = "subprocess.communicate" TEMPLATE_RENDER = "template.render" VIEW_RENDER = "view.render" VIEW_RESPONSE_RENDER = "view.response.render" WEBSOCKET_SERVER = "websocket.server" SOCKET_CONNECTION = "socket.connection" SOCKET_DNS = "socket.dns" # This type exists to trick mypy and PyCharm into thinking `init` and `Client` # take these arguments (even though they take opaque *kwargs) class ClientConstructor(object): def __init__( self, dsn=None, # type: Optional[str] max_breadcrumbs=DEFAULT_MAX_BREADCRUMBS, # type: int release=None, # type: Optional[str] environment=None, # type: Optional[str] server_name=None, # type: Optional[str] shutdown_timeout=2, # type: float integrations=[], # type: Sequence[Integration] # noqa: B006 in_app_include=[], # type: List[str] # noqa: B006 in_app_exclude=[], # type: List[str] # noqa: B006 default_integrations=True, # type: bool dist=None, # type: Optional[str] transport=None, # type: Optional[Union[sentry_sdk.transport.Transport, Type[sentry_sdk.transport.Transport], Callable[[Event], None]]] transport_queue_size=DEFAULT_QUEUE_SIZE, # type: int sample_rate=1.0, # type: float send_default_pii=False, # type: bool http_proxy=None, # type: Optional[str] https_proxy=None, # type: Optional[str] ignore_errors=[], # type: Sequence[Union[type, str]] # noqa: B006 max_request_body_size="medium", # type: str before_send=None, # type: Optional[EventProcessor] before_breadcrumb=None, # type: Optional[BreadcrumbProcessor] debug=False, # type: bool attach_stacktrace=False, # type: bool ca_certs=None, # type: Optional[str] propagate_traces=True, # type: bool traces_sample_rate=None, # type: Optional[float] traces_sampler=None, # type: Optional[TracesSampler] profiles_sample_rate=None, # type: Optional[float] profiles_sampler=None, # type: Optional[TracesSampler] profiler_mode=None, # type: Optional[ProfilerMode] auto_enabling_integrations=True, # type: bool auto_session_tracking=True, # type: bool send_client_reports=True, # type: bool _experiments={}, # type: Experiments # noqa: B006 proxy_headers=None, # type: Optional[Dict[str, str]] instrumenter=INSTRUMENTER.SENTRY, # type: Optional[str] before_send_transaction=None, # type: Optional[TransactionProcessor] project_root=None, # type: Optional[str] enable_tracing=None, # type: Optional[bool] include_local_variables=True, # type: Optional[bool] include_source_context=True, # type: Optional[bool] trace_propagation_targets=[ # noqa: B006 MATCH_ALL ], # type: Optional[Sequence[str]] functions_to_trace=[], # type: Sequence[Dict[str, str]] # noqa: B006 event_scrubber=None, # type: Optional[sentry_sdk.scrubber.EventScrubber] max_value_length=DEFAULT_MAX_VALUE_LENGTH, # type: int ): # type: (...) -> None pass def _get_default_options(): # type: () -> Dict[str, Any] import inspect if hasattr(inspect, "getfullargspec"): getargspec = inspect.getfullargspec else: getargspec = inspect.getargspec # type: ignore a = getargspec(ClientConstructor.__init__) defaults = a.defaults or () return dict(zip(a.args[-len(defaults) :], defaults)) DEFAULT_OPTIONS = _get_default_options() del _get_default_options VERSION = "1.29.2"	8.49 KB	2026-01-20 12:58
📁 crons	-	2026-03-21 15:45
📄 debug.py import sys import logging from sentry_sdk import utils from sentry_sdk.hub import Hub from sentry_sdk.utils import logger from sentry_sdk.client import _client_init_debug from logging import LogRecord class _HubBasedClientFilter(logging.Filter): def filter(self, record): # type: (LogRecord) -> bool if _client_init_debug.get(False): return True hub = Hub.current if hub is not None and hub.client is not None: return hub.client.options["debug"] return False def init_debug_support(): # type: () -> None if not logger.handlers: configure_logger() configure_debug_hub() def configure_logger(): # type: () -> None _handler = logging.StreamHandler(sys.stderr) _handler.setFormatter(logging.Formatter(" [sentry] %(levelname)s: %(message)s")) logger.addHandler(_handler) logger.setLevel(logging.DEBUG) logger.addFilter(_HubBasedClientFilter()) def configure_debug_hub(): # type: () -> None def _get_debug_hub(): # type: () -> Hub return Hub.current utils._get_debug_hub = _get_debug_hub	1.11 KB	2026-01-20 12:58
📄 envelope.py import io import json import mimetypes from sentry_sdk._compat import text_type, PY2 from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.session import Session from sentry_sdk.utils import json_dumps, capture_internal_exceptions if TYPE_CHECKING: from typing import Any from typing import Optional from typing import Union from typing import Dict from typing import List from typing import Iterator from sentry_sdk._types import Event, EventDataCategory def parse_json(data): # type: (Union[bytes, text_type]) -> Any # on some python 3 versions this needs to be bytes if not PY2 and isinstance(data, bytes): data = data.decode("utf-8", "replace") return json.loads(data) class Envelope(object): def __init__( self, headers=None, # type: Optional[Dict[str, Any]] items=None, # type: Optional[List[Item]] ): # type: (...) -> None if headers is not None: headers = dict(headers) self.headers = headers or {} if items is None: items = [] else: items = list(items) self.items = items @property def description(self): # type: (...) -> str return "envelope with %s items (%s)" % ( len(self.items), ", ".join(x.data_category for x in self.items), ) def add_event( self, event # type: Event ): # type: (...) -> None self.add_item(Item(payload=PayloadRef(json=event), type="event")) def add_transaction( self, transaction # type: Event ): # type: (...) -> None self.add_item(Item(payload=PayloadRef(json=transaction), type="transaction")) def add_profile( self, profile # type: Any ): # type: (...) -> None self.add_item(Item(payload=PayloadRef(json=profile), type="profile")) def add_checkin( self, checkin # type: Any ): # type: (...) -> None self.add_item(Item(payload=PayloadRef(json=checkin), type="check_in")) def add_session( self, session # type: Union[Session, Any] ): # type: (...) -> None if isinstance(session, Session): session = session.to_json() self.add_item(Item(payload=PayloadRef(json=session), type="session")) def add_sessions( self, sessions # type: Any ): # type: (...) -> None self.add_item(Item(payload=PayloadRef(json=sessions), type="sessions")) def add_item( self, item # type: Item ): # type: (...) -> None self.items.append(item) def get_event(self): # type: (...) -> Optional[Event] for items in self.items: event = items.get_event() if event is not None: return event return None def get_transaction_event(self): # type: (...) -> Optional[Event] for item in self.items: event = item.get_transaction_event() if event is not None: return event return None def __iter__(self): # type: (...) -> Iterator[Item] return iter(self.items) def serialize_into( self, f # type: Any ): # type: (...) -> None f.write(json_dumps(self.headers)) f.write(b"\n") for item in self.items: item.serialize_into(f) def serialize(self): # type: (...) -> bytes out = io.BytesIO() self.serialize_into(out) return out.getvalue() @classmethod def deserialize_from( cls, f # type: Any ): # type: (...) -> Envelope headers = parse_json(f.readline()) items = [] while 1: item = Item.deserialize_from(f) if item is None: break items.append(item) return cls(headers=headers, items=items) @classmethod def deserialize( cls, bytes # type: bytes ): # type: (...) -> Envelope return cls.deserialize_from(io.BytesIO(bytes)) def __repr__(self): # type: (...) -> str return "<Envelope headers=%r items=%r>" % (self.headers, self.items) class PayloadRef(object): def __init__( self, bytes=None, # type: Optional[bytes] path=None, # type: Optional[Union[bytes, text_type]] json=None, # type: Optional[Any] ): # type: (...) -> None self.json = json self.bytes = bytes self.path = path def get_bytes(self): # type: (...) -> bytes if self.bytes is None: if self.path is not None: with capture_internal_exceptions(): with open(self.path, "rb") as f: self.bytes = f.read() elif self.json is not None: self.bytes = json_dumps(self.json) else: self.bytes = b"" return self.bytes @property def inferred_content_type(self): # type: (...) -> str if self.json is not None: return "application/json" elif self.path is not None: path = self.path if isinstance(path, bytes): path = path.decode("utf-8", "replace") ty = mimetypes.guess_type(path)[0] if ty: return ty return "application/octet-stream" def __repr__(self): # type: (...) -> str return "<Payload %r>" % (self.inferred_content_type,) class Item(object): def __init__( self, payload, # type: Union[bytes, text_type, PayloadRef] headers=None, # type: Optional[Dict[str, Any]] type=None, # type: Optional[str] content_type=None, # type: Optional[str] filename=None, # type: Optional[str] ): if headers is not None: headers = dict(headers) elif headers is None: headers = {} self.headers = headers if isinstance(payload, bytes): payload = PayloadRef(bytes=payload) elif isinstance(payload, text_type): payload = PayloadRef(bytes=payload.encode("utf-8")) else: payload = payload if filename is not None: headers["filename"] = filename if type is not None: headers["type"] = type if content_type is not None: headers["content_type"] = content_type elif "content_type" not in headers: headers["content_type"] = payload.inferred_content_type self.payload = payload def __repr__(self): # type: (...) -> str return "<Item headers=%r payload=%r data_category=%r>" % ( self.headers, self.payload, self.data_category, ) @property def type(self): # type: (...) -> Optional[str] return self.headers.get("type") @property def data_category(self): # type: (...) -> EventDataCategory ty = self.headers.get("type") if ty == "session": return "session" elif ty == "attachment": return "attachment" elif ty == "transaction": return "transaction" elif ty == "event": return "error" elif ty == "client_report": return "internal" elif ty == "profile": return "profile" else: return "default" def get_bytes(self): # type: (...) -> bytes return self.payload.get_bytes() def get_event(self): # type: (...) -> Optional[Event] """ Returns an error event if there is one. """ if self.type == "event" and self.payload.json is not None: return self.payload.json return None def get_transaction_event(self): # type: (...) -> Optional[Event] if self.type == "transaction" and self.payload.json is not None: return self.payload.json return None def serialize_into( self, f # type: Any ): # type: (...) -> None headers = dict(self.headers) bytes = self.get_bytes() headers["length"] = len(bytes) f.write(json_dumps(headers)) f.write(b"\n") f.write(bytes) f.write(b"\n") def serialize(self): # type: (...) -> bytes out = io.BytesIO() self.serialize_into(out) return out.getvalue() @classmethod def deserialize_from( cls, f # type: Any ): # type: (...) -> Optional[Item] line = f.readline().rstrip() if not line: return None headers = parse_json(line) length = headers.get("length") if length is not None: payload = f.read(length) f.readline() else: # if no length was specified we need to read up to the end of line # and remove it (if it is present, i.e. not the very last char in an eof terminated envelope) payload = f.readline().rstrip(b"\n") if headers.get("type") in ("event", "transaction", "metric_buckets"): rv = cls(headers=headers, payload=PayloadRef(json=parse_json(payload))) else: rv = cls(headers=headers, payload=payload) return rv @classmethod def deserialize( cls, bytes # type: bytes ): # type: (...) -> Optional[Item] return cls.deserialize_from(io.BytesIO(bytes))	9.37 KB	2026-01-20 12:58
📄 hub.py import copy import sys from datetime import datetime from contextlib import contextmanager from sentry_sdk._compat import with_metaclass from sentry_sdk.consts import INSTRUMENTER from sentry_sdk.scope import Scope from sentry_sdk.client import Client from sentry_sdk.profiler import Profile from sentry_sdk.tracing import ( NoOpSpan, Span, Transaction, BAGGAGE_HEADER_NAME, SENTRY_TRACE_HEADER_NAME, ) from sentry_sdk.session import Session from sentry_sdk.tracing_utils import ( has_tracing_enabled, normalize_incoming_data, ) from sentry_sdk.utils import ( exc_info_from_error, event_from_exception, logger, ContextVar, ) from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Union from typing import Any from typing import Optional from typing import Tuple from typing import Dict from typing import List from typing import Callable from typing import Generator from typing import Type from typing import TypeVar from typing import overload from typing import ContextManager from sentry_sdk.integrations import Integration from sentry_sdk._types import ( Event, Hint, Breadcrumb, BreadcrumbHint, ExcInfo, ) from sentry_sdk.consts import ClientConstructor T = TypeVar("T") else: def overload(x): # type: (T) -> T return x _local = ContextVar("sentry_current_hub") def _update_scope(base, scope_change, scope_kwargs): # type: (Scope, Optional[Any], Dict[str, Any]) -> Scope if scope_change and scope_kwargs: raise TypeError("cannot provide scope and kwargs") if scope_change is not None: final_scope = copy.copy(base) if callable(scope_change): scope_change(final_scope) else: final_scope.update_from_scope(scope_change) elif scope_kwargs: final_scope = copy.copy(base) final_scope.update_from_kwargs(*scope_kwargs) else: final_scope = base return final_scope def _should_send_default_pii(): # type: () -> bool client = Hub.current.client if not client: return False return client.options["send_default_pii"] class _InitGuard(object): def __init__(self, client): # type: (Client) -> None self._client = client def __enter__(self): # type: () -> _InitGuard return self def __exit__(self, exc_type, exc_value, tb): # type: (Any, Any, Any) -> None c = self._client if c is not None: c.close() def _check_python_deprecations(): # type: () -> None version = sys.version_info[:2] if version == (3, 4) or version == (3, 5): logger.warning( "sentry-sdk 2.0.0 will drop support for Python %s.", "{}.{}".format(version), ) logger.warning( "Please upgrade to the latest version to continue receiving upgrades and bugfixes." ) def _init(args, kwargs): # type: (Optional[str], *Any) -> ContextManager[Any] """Initializes the SDK and optionally integrations. This takes the same arguments as the client constructor. """ client = Client(args, kwargs) # type: ignore Hub.current.bind_client(client) _check_python_deprecations() rv = _InitGuard(client) return rv from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: # Make mypy, PyCharm and other static analyzers think `init` is a type to # have nicer autocompletion for params. # # Use `ClientConstructor` to define the argument types of `init` and # `ContextManager[Any]` to tell static analyzers about the return type. class init(ClientConstructor, _InitGuard): # noqa: N801 pass else: # Alias `init` for actual usage. Go through the lambda indirection to throw # PyCharm off of the weakly typed signature (it would otherwise discover # both the weakly typed signature of `_init` and our faked `init` type). init = (lambda: _init)() class HubMeta(type): @property def current(cls): # type: () -> Hub """Returns the current instance of the hub.""" rv = _local.get(None) if rv is None: rv = Hub(GLOBAL_HUB) _local.set(rv) return rv @property def main(cls): # type: () -> Hub """Returns the main instance of the hub.""" return GLOBAL_HUB class _ScopeManager(object): def __init__(self, hub): # type: (Hub) -> None self._hub = hub self._original_len = len(hub._stack) self._layer = hub._stack[-1] def __enter__(self): # type: () -> Scope scope = self._layer[1] assert scope is not None return scope def __exit__(self, exc_type, exc_value, tb): # type: (Any, Any, Any) -> None current_len = len(self._hub._stack) if current_len < self._original_len: logger.error( "Scope popped too soon. Popped %s scopes too many.", self._original_len - current_len, ) return elif current_len > self._original_len: logger.warning( "Leaked %s scopes: %s", current_len - self._original_len, self._hub._stack[self._original_len :], ) layer = self._hub._stack[self._original_len - 1] del self._hub._stack[self._original_len - 1 :] if layer[1] != self._layer[1]: logger.error( "Wrong scope found. Meant to pop %s, but popped %s.", layer[1], self._layer[1], ) elif layer[0] != self._layer[0]: warning = ( "init() called inside of pushed scope. This might be entirely " "legitimate but usually occurs when initializing the SDK inside " "a request handler or task/job function. Try to initialize the " "SDK as early as possible instead." ) logger.warning(warning) class Hub(with_metaclass(HubMeta)): # type: ignore """The hub wraps the concurrency management of the SDK. Each thread has its own hub but the hub might transfer with the flow of execution if context vars are available. If the hub is used with a with statement it's temporarily activated. """ _stack = None # type: List[Tuple[Optional[Client], Scope]] # Mypy doesn't pick up on the metaclass. if TYPE_CHECKING: current = None # type: Hub main = None # type: Hub def __init__( self, client_or_hub=None, # type: Optional[Union[Hub, Client]] scope=None, # type: Optional[Any] ): # type: (...) -> None if isinstance(client_or_hub, Hub): hub = client_or_hub client, other_scope = hub._stack[-1] if scope is None: scope = copy.copy(other_scope) else: client = client_or_hub if scope is None: scope = Scope() self._stack = [(client, scope)] self._last_event_id = None # type: Optional[str] self._old_hubs = [] # type: List[Hub] def __enter__(self): # type: () -> Hub self._old_hubs.append(Hub.current) _local.set(self) return self def __exit__( self, exc_type, # type: Optional[type] exc_value, # type: Optional[BaseException] tb, # type: Optional[Any] ): # type: (...) -> None old = self._old_hubs.pop() _local.set(old) def run( self, callback # type: Callable[[], T] ): # type: (...) -> T """Runs a callback in the context of the hub. Alternatively the with statement can be used on the hub directly. """ with self: return callback() def get_integration( self, name_or_class # type: Union[str, Type[Integration]] ): # type: (...) -> Any """Returns the integration for this hub by name or class. If there is no client bound or the client does not have that integration then `None` is returned. If the return value is not `None` the hub is guaranteed to have a client attached. """ if isinstance(name_or_class, str): integration_name = name_or_class elif name_or_class.identifier is not None: integration_name = name_or_class.identifier else: raise ValueError("Integration has no name") client = self.client if client is not None: rv = client.integrations.get(integration_name) if rv is not None: return rv @property def client(self): # type: () -> Optional[Client] """Returns the current client on the hub.""" return self._stack[-1][0] @property def scope(self): # type: () -> Scope """Returns the current scope on the hub.""" return self._stack[-1][1] def last_event_id(self): # type: () -> Optional[str] """Returns the last event ID.""" return self._last_event_id def bind_client( self, new # type: Optional[Client] ): # type: (...) -> None """Binds a new client to the hub.""" top = self._stack[-1] self._stack[-1] = (new, top[1]) def capture_event(self, event, hint=None, scope=None, scope_args): # type: (Event, Optional[Hint], Optional[Scope], Any) -> Optional[str] """ Captures an event. Alias of :py:meth:`sentry_sdk.Client.capture_event`. :param scope_args: For supported `scope_args` see :py:meth:`sentry_sdk.Scope.update_from_kwargs`. """ client, top_scope = self._stack[-1] scope = _update_scope(top_scope, scope, scope_args) if client is not None: is_transaction = event.get("type") == "transaction" rv = client.capture_event(event, hint, scope) if rv is not None and not is_transaction: self._last_event_id = rv return rv return None def capture_message(self, message, level=None, scope=None, scope_args): # type: (str, Optional[str], Optional[Scope], Any) -> Optional[str] """ Captures a message. :param message: The string to send as the message. :param level: If no level is provided, the default level is `info`. :param scope: An optional :py:class:`sentry_sdk.Scope` to use. :param scope_args: For supported `scope_args` see :py:meth:`sentry_sdk.Scope.update_from_kwargs`. :returns: An `event_id` if the SDK decided to send the event (see :py:meth:`sentry_sdk.Client.capture_event`). """ if self.client is None: return None if level is None: level = "info" return self.capture_event( {"message": message, "level": level}, scope=scope, scope_args ) def capture_exception(self, error=None, scope=None, scope_args): # type: (Optional[Union[BaseException, ExcInfo]], Optional[Scope], Any) -> Optional[str] """Captures an exception. :param error: An exception to catch. If `None`, `sys.exc_info()` will be used. :param scope_args: For supported `scope_args` see :py:meth:`sentry_sdk.Scope.update_from_kwargs`. :returns: An `event_id` if the SDK decided to send the event (see :py:meth:`sentry_sdk.Client.capture_event`). """ client = self.client if client is None: return None if error is not None: exc_info = exc_info_from_error(error) else: exc_info = sys.exc_info() event, hint = event_from_exception(exc_info, client_options=client.options) try: return self.capture_event(event, hint=hint, scope=scope, scope_args) except Exception: self._capture_internal_exception(sys.exc_info()) return None def _capture_internal_exception( self, exc_info # type: Any ): # type: (...) -> Any """ Capture an exception that is likely caused by a bug in the SDK itself. These exceptions do not end up in Sentry and are just logged instead. """ logger.error("Internal error in sentry_sdk", exc_info=exc_info) def add_breadcrumb(self, crumb=None, hint=None, kwargs): # type: (Optional[Breadcrumb], Optional[BreadcrumbHint], Any) -> None """ Adds a breadcrumb. :param crumb: Dictionary with the data as the sentry v7/v8 protocol expects. :param hint: An optional value that can be used by `before_breadcrumb` to customize the breadcrumbs that are emitted. """ client, scope = self._stack[-1] if client is None: logger.info("Dropped breadcrumb because no client bound") return crumb = dict(crumb or ()) # type: Breadcrumb crumb.update(kwargs) if not crumb: return hint = dict(hint or ()) # type: Hint if crumb.get("timestamp") is None: crumb["timestamp"] = datetime.utcnow() if crumb.get("type") is None: crumb["type"] = "default" if client.options["before_breadcrumb"] is not None: new_crumb = client.options["before_breadcrumb"](crumb, hint) else: new_crumb = crumb if new_crumb is not None: scope._breadcrumbs.append(new_crumb) else: logger.info("before breadcrumb dropped breadcrumb (%s)", crumb) max_breadcrumbs = client.options["max_breadcrumbs"] # type: int while len(scope._breadcrumbs) > max_breadcrumbs: scope._breadcrumbs.popleft() def start_span(self, span=None, instrumenter=INSTRUMENTER.SENTRY, kwargs): # type: (Optional[Span], str, Any) -> Span """ Start a span whose parent is the currently active span or transaction, if any. The return value is a :py:class:`sentry_sdk.tracing.Span` instance, typically used as a context manager to start and stop timing in a `with` block. Only spans contained in a transaction are sent to Sentry. Most integrations start a transaction at the appropriate time, for example for every incoming HTTP request. Use :py:meth:`sentry_sdk.start_transaction` to start a new transaction when one is not already in progress. For supported `kwargs` see :py:class:`sentry_sdk.tracing.Span`. """ configuration_instrumenter = self.client and self.client.options["instrumenter"] if instrumenter != configuration_instrumenter: return NoOpSpan() # TODO: consider removing this in a future release. # This is for backwards compatibility with releases before # start_transaction existed, to allow for a smoother transition. if isinstance(span, Transaction) or "transaction" in kwargs: deprecation_msg = ( "Deprecated: use start_transaction to start transactions and " "Transaction.start_child to start spans." ) if isinstance(span, Transaction): logger.warning(deprecation_msg) return self.start_transaction(span) if "transaction" in kwargs: logger.warning(deprecation_msg) name = kwargs.pop("transaction") return self.start_transaction(name=name, kwargs) if span is not None: return span kwargs.setdefault("hub", self) span = self.scope.span if span is not None: return span.start_child(kwargs) # If there is already a trace_id in the propagation context, use it. if "trace_id" not in kwargs: traceparent = self.get_traceparent() trace_id = traceparent.split("-")[0] if traceparent else None if trace_id is not None: kwargs["trace_id"] = trace_id return Span(kwargs) def start_transaction( self, transaction=None, instrumenter=INSTRUMENTER.SENTRY, kwargs ): # type: (Optional[Transaction], str, Any) -> Union[Transaction, NoOpSpan] """ Start and return a transaction. Start an existing transaction if given, otherwise create and start a new transaction with kwargs. This is the entry point to manual tracing instrumentation. A tree structure can be built by adding child spans to the transaction, and child spans to other spans. To start a new child span within the transaction or any span, call the respective `.start_child()` method. Every child span must be finished before the transaction is finished, otherwise the unfinished spans are discarded. When used as context managers, spans and transactions are automatically finished at the end of the `with` block. If not using context managers, call the `.finish()` method. When the transaction is finished, it will be sent to Sentry with all its finished child spans. For supported `kwargs` see :py:class:`sentry_sdk.tracing.Transaction`. """ configuration_instrumenter = self.client and self.client.options["instrumenter"] if instrumenter != configuration_instrumenter: return NoOpSpan() custom_sampling_context = kwargs.pop("custom_sampling_context", {}) # if we haven't been given a transaction, make one if transaction is None: kwargs.setdefault("hub", self) transaction = Transaction(kwargs) # use traces_sample_rate, traces_sampler, and/or inheritance to make a # sampling decision sampling_context = { "transaction_context": transaction.to_json(), "parent_sampled": transaction.parent_sampled, } sampling_context.update(custom_sampling_context) transaction._set_initial_sampling_decision(sampling_context=sampling_context) profile = Profile(transaction, hub=self) profile._set_initial_sampling_decision(sampling_context=sampling_context) # we don't bother to keep spans if we already know we're not going to # send the transaction if transaction.sampled: max_spans = ( self.client and self.client.options["_experiments"].get("max_spans") ) or 1000 transaction.init_span_recorder(maxlen=max_spans) return transaction def continue_trace(self, environ_or_headers, op=None, name=None, source=None): # type: (Dict[str, Any], Optional[str], Optional[str], Optional[str]) -> Transaction """ Sets the propagation context from environment or headers and returns a transaction. """ with self.configure_scope() as scope: scope.generate_propagation_context(environ_or_headers) transaction = Transaction.continue_from_headers( normalize_incoming_data(environ_or_headers), op=op, name=name, source=source, ) return transaction @overload def push_scope( self, callback=None # type: Optional[None] ): # type: (...) -> ContextManager[Scope] pass @overload def push_scope( # noqa: F811 self, callback # type: Callable[[Scope], None] ): # type: (...) -> None pass def push_scope( # noqa self, callback=None, # type: Optional[Callable[[Scope], None]] continue_trace=True, # type: bool ): # type: (...) -> Optional[ContextManager[Scope]] """ Pushes a new layer on the scope stack. :param callback: If provided, this method pushes a scope, calls `callback`, and pops the scope again. :returns: If no `callback` is provided, a context manager that should be used to pop the scope again. """ if callback is not None: with self.push_scope() as scope: callback(scope) return None client, scope = self._stack[-1] new_scope = copy.copy(scope) if continue_trace: new_scope.generate_propagation_context() new_layer = (client, new_scope) self._stack.append(new_layer) return _ScopeManager(self) def pop_scope_unsafe(self): # type: () -> Tuple[Optional[Client], Scope] """ Pops a scope layer from the stack. Try to use the context manager :py:meth:`push_scope` instead. """ rv = self._stack.pop() assert self._stack, "stack must have at least one layer" return rv @overload def configure_scope( self, callback=None # type: Optional[None] ): # type: (...) -> ContextManager[Scope] pass @overload def configure_scope( # noqa: F811 self, callback # type: Callable[[Scope], None] ): # type: (...) -> None pass def configure_scope( # noqa self, callback=None, # type: Optional[Callable[[Scope], None]] continue_trace=True, # type: bool ): # type: (...) -> Optional[ContextManager[Scope]] """ Reconfigures the scope. :param callback: If provided, call the callback with the current scope. :returns: If no callback is provided, returns a context manager that returns the scope. """ client, scope = self._stack[-1] if continue_trace: scope.generate_propagation_context() if callback is not None: if client is not None: callback(scope) return None @contextmanager def inner(): # type: () -> Generator[Scope, None, None] if client is not None: yield scope else: yield Scope() return inner() def start_session( self, session_mode="application" # type: str ): # type: (...) -> None """Starts a new session.""" self.end_session() client, scope = self._stack[-1] scope._session = Session( release=client.options["release"] if client else None, environment=client.options["environment"] if client else None, user=scope._user, session_mode=session_mode, ) def end_session(self): # type: (...) -> None """Ends the current session if there is one.""" client, scope = self._stack[-1] session = scope._session self.scope._session = None if session is not None: session.close() if client is not None: client.capture_session(session) def stop_auto_session_tracking(self): # type: (...) -> None """Stops automatic session tracking. This temporarily session tracking for the current scope when called. To resume session tracking call `resume_auto_session_tracking`. """ self.end_session() client, scope = self._stack[-1] scope._force_auto_session_tracking = False def resume_auto_session_tracking(self): # type: (...) -> None """Resumes automatic session tracking for the current scope if disabled earlier. This requires that generally automatic session tracking is enabled. """ client, scope = self._stack[-1] scope._force_auto_session_tracking = None def flush( self, timeout=None, # type: Optional[float] callback=None, # type: Optional[Callable[[int, float], None]] ): # type: (...) -> None """ Alias for :py:meth:`sentry_sdk.Client.flush` """ client, scope = self._stack[-1] if client is not None: return client.flush(timeout=timeout, callback=callback) def get_traceparent(self): # type: () -> Optional[str] """ Returns the traceparent either from the active span or from the scope. """ if self.client is not None: if has_tracing_enabled(self.client.options) and self.scope.span is not None: return self.scope.span.to_traceparent() return self.scope.get_traceparent() def get_baggage(self): # type: () -> Optional[str] """ Returns Baggage either from the active span or from the scope. """ if ( self.client is not None and has_tracing_enabled(self.client.options) and self.scope.span is not None ): baggage = self.scope.span.to_baggage() else: baggage = self.scope.get_baggage() if baggage is not None: return baggage.serialize() return None def iter_trace_propagation_headers(self, span=None): # type: (Optional[Span]) -> Generator[Tuple[str, str], None, None] """ Return HTTP headers which allow propagation of trace data. Data taken from the span representing the request, if available, or the current span on the scope if not. """ client = self._stack[-1][0] propagate_traces = client and client.options["propagate_traces"] if not propagate_traces: return span = span or self.scope.span if client and has_tracing_enabled(client.options) and span is not None: for header in span.iter_headers(): yield header else: for header in self.scope.iter_headers(): yield header def trace_propagation_meta(self, span=None): # type: (Optional[Span]) -> str """ Return meta tags which should be injected into HTML templates to allow propagation of trace information. """ if span is not None: logger.warning( "The parameter `span` in trace_propagation_meta() is deprecated and will be removed in the future." ) meta = "" sentry_trace = self.get_traceparent() if sentry_trace is not None: meta += '<meta name="%s" content="%s">' % ( SENTRY_TRACE_HEADER_NAME, sentry_trace, ) baggage = self.get_baggage() if baggage is not None: meta += '<meta name="%s" content="%s">' % ( BAGGAGE_HEADER_NAME, baggage, ) return meta GLOBAL_HUB = Hub() _local.set(GLOBAL_HUB)	26.42 KB	2026-01-20 12:58
📁 integrations	-	2026-03-21 15:45
📄 monitor.py import os import time from threading import Thread, Lock import sentry_sdk from sentry_sdk.utils import logger from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Optional class Monitor(object): """ Performs health checks in a separate thread once every interval seconds and updates the internal state. Other parts of the SDK only read this state and act accordingly. """ name = "sentry.monitor" def __init__(self, transport, interval=10): # type: (sentry_sdk.transport.Transport, float) -> None self.transport = transport # type: sentry_sdk.transport.Transport self.interval = interval # type: float self._healthy = True self._downsample_factor = 1 # type: int self._thread = None # type: Optional[Thread] self._thread_lock = Lock() self._thread_for_pid = None # type: Optional[int] self._running = True def _ensure_running(self): # type: () -> None if self._thread_for_pid == os.getpid() and self._thread is not None: return None with self._thread_lock: if self._thread_for_pid == os.getpid() and self._thread is not None: return None def _thread(): # type: (...) -> None while self._running: time.sleep(self.interval) if self._running: self.run() thread = Thread(name=self.name, target=_thread) thread.daemon = True thread.start() self._thread = thread self._thread_for_pid = os.getpid() return None def run(self): # type: () -> None self.check_health() self.set_downsample_factor() def set_downsample_factor(self): # type: () -> None if self._healthy: if self._downsample_factor > 1: logger.debug( "[Monitor] health check positive, reverting to normal sampling" ) self._downsample_factor = 1 else: self._downsample_factor *= 2 logger.debug( "[Monitor] health check negative, downsampling with a factor of %d", self._downsample_factor, ) def check_health(self): # type: () -> None """ Perform the actual health checks, currently only checks if the transport is rate-limited. TODO: augment in the future with more checks. """ self._healthy = self.transport.is_healthy() def is_healthy(self): # type: () -> bool self._ensure_running() return self._healthy @property def downsample_factor(self): # type: () -> int self._ensure_running() return self._downsample_factor def kill(self): # type: () -> None self._running = False def __del__(self): # type: () -> None self.kill()	2.97 KB	2026-01-20 12:58
📄 profiler.py """ This file is originally based on code from https://github.com/nylas/nylas-perftools, which is published under the following license: The MIT License (MIT) Copyright (c) 2014 Nylas Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. """ import atexit import os import platform import random import sys import threading import time import uuid from collections import deque import sentry_sdk from sentry_sdk._compat import PY33, PY311 from sentry_sdk._lru_cache import LRUCache from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.utils import ( capture_internal_exception, filename_for_module, is_valid_sample_rate, logger, nanosecond_time, set_in_app_in_frames, ) if TYPE_CHECKING: from types import FrameType from typing import Any from typing import Callable from typing import Deque from typing import Dict from typing import List from typing import Optional from typing import Set from typing import Sequence from typing import Tuple from typing_extensions import TypedDict import sentry_sdk.tracing from sentry_sdk._types import SamplingContext, ProfilerMode ThreadId = str ProcessedSample = TypedDict( "ProcessedSample", { "elapsed_since_start_ns": str, "thread_id": ThreadId, "stack_id": int, }, ) ProcessedStack = List[int] ProcessedFrame = TypedDict( "ProcessedFrame", { "abs_path": str, "filename": Optional[str], "function": str, "lineno": int, "module": Optional[str], }, ) ProcessedThreadMetadata = TypedDict( "ProcessedThreadMetadata", {"name": str}, ) ProcessedProfile = TypedDict( "ProcessedProfile", { "frames": List[ProcessedFrame], "stacks": List[ProcessedStack], "samples": List[ProcessedSample], "thread_metadata": Dict[ThreadId, ProcessedThreadMetadata], }, ) ProfileContext = TypedDict( "ProfileContext", {"profile_id": str}, ) FrameId = Tuple[ str, # abs_path int, # lineno str, # function ] FrameIds = Tuple[FrameId, ...] # The exact value of this id is not very meaningful. The purpose # of this id is to give us a compact and unique identifier for a # raw stack that can be used as a key to a dictionary so that it # can be used during the sampled format generation. StackId = Tuple[int, int] ExtractedStack = Tuple[StackId, FrameIds, List[ProcessedFrame]] ExtractedSample = Sequence[Tuple[ThreadId, ExtractedStack]] try: from gevent import get_hub as get_gevent_hub # type: ignore from gevent.monkey import get_original, is_module_patched # type: ignore from gevent.threadpool import ThreadPool # type: ignore thread_sleep = get_original("time", "sleep") except ImportError: def get_gevent_hub(): # type: () -> Any return None thread_sleep = time.sleep def is_module_patched(args, kwargs): # type: (Any, *Any) -> bool # unable to import from gevent means no modules have been patched return False ThreadPool = None def is_gevent(): # type: () -> bool return is_module_patched("threading") or is_module_patched("_thread") _scheduler = None # type: Optional[Scheduler] # The default sampling frequency to use. This is set at 101 in order to # mitigate the effects of lockstep sampling. DEFAULT_SAMPLING_FREQUENCY = 101 # The minimum number of unique samples that must exist in a profile to be # considered valid. PROFILE_MINIMUM_SAMPLES = 2 def has_profiling_enabled(options): # type: (Dict[str, Any]) -> bool profiles_sampler = options["profiles_sampler"] if profiles_sampler is not None: return True profiles_sample_rate = options["profiles_sample_rate"] if profiles_sample_rate is not None and profiles_sample_rate > 0: return True profiles_sample_rate = options["_experiments"].get("profiles_sample_rate") if profiles_sample_rate is not None and profiles_sample_rate > 0: return True return False def setup_profiler(options): # type: (Dict[str, Any]) -> bool global _scheduler if _scheduler is not None: logger.debug("[Profiling] Profiler is already setup") return False if not PY33: logger.warn("[Profiling] Profiler requires Python >= 3.3") return False frequency = DEFAULT_SAMPLING_FREQUENCY if is_gevent(): # If gevent has patched the threading modules then we cannot rely on # them to spawn a native thread for sampling. # Instead we default to the GeventScheduler which is capable of # spawning native threads within gevent. default_profiler_mode = GeventScheduler.mode else: default_profiler_mode = ThreadScheduler.mode if options.get("profiler_mode") is not None: profiler_mode = options["profiler_mode"] else: profiler_mode = ( options.get("_experiments", {}).get("profiler_mode") or default_profiler_mode ) if ( profiler_mode == ThreadScheduler.mode # for legacy reasons, we'll keep supporting sleep mode for this scheduler or profiler_mode == "sleep" ): _scheduler = ThreadScheduler(frequency=frequency) elif profiler_mode == GeventScheduler.mode: _scheduler = GeventScheduler(frequency=frequency) else: raise ValueError("Unknown profiler mode: {}".format(profiler_mode)) logger.debug( "[Profiling] Setting up profiler in {mode} mode".format(mode=_scheduler.mode) ) _scheduler.setup() atexit.register(teardown_profiler) return True def teardown_profiler(): # type: () -> None global _scheduler if _scheduler is not None: _scheduler.teardown() _scheduler = None # We want to impose a stack depth limit so that samples aren't too large. MAX_STACK_DEPTH = 128 CWD = os.getcwd() def extract_stack( raw_frame, # type: Optional[FrameType] cache, # type: LRUCache cwd=CWD, # type: str max_stack_depth=MAX_STACK_DEPTH, # type: int ): # type: (...) -> ExtractedStack """ Extracts the stack starting the specified frame. The extracted stack assumes the specified frame is the top of the stack, and works back to the bottom of the stack. In the event that the stack is more than `MAX_STACK_DEPTH` frames deep, only the first `MAX_STACK_DEPTH` frames will be returned. """ raw_frames = deque(maxlen=max_stack_depth) # type: Deque[FrameType] while raw_frame is not None: f_back = raw_frame.f_back raw_frames.append(raw_frame) raw_frame = f_back frame_ids = tuple(frame_id(raw_frame) for raw_frame in raw_frames) frames = [] for i, fid in enumerate(frame_ids): frame = cache.get(fid) if frame is None: frame = extract_frame(fid, raw_frames[i], cwd) cache.set(fid, frame) frames.append(frame) # Instead of mapping the stack into frame ids and hashing # that as a tuple, we can directly hash the stack. # This saves us from having to generate yet another list. # Additionally, using the stack as the key directly is # costly because the stack can be large, so we pre-hash # the stack, and use the hash as the key as this will be # needed a few times to improve performance. # # To Reduce the likelihood of hash collisions, we include # the stack depth. This means that only stacks of the same # depth can suffer from hash collisions. stack_id = len(raw_frames), hash(frame_ids) return stack_id, frame_ids, frames def frame_id(raw_frame): # type: (FrameType) -> FrameId return (raw_frame.f_code.co_filename, raw_frame.f_lineno, get_frame_name(raw_frame)) def extract_frame(fid, raw_frame, cwd): # type: (FrameId, FrameType, str) -> ProcessedFrame abs_path = raw_frame.f_code.co_filename try: module = raw_frame.f_globals["__name__"] except Exception: module = None # namedtuples can be many times slower when initialing # and accessing attribute so we opt to use a tuple here instead return { # This originally was `os.path.abspath(abs_path)` but that had # a large performance overhead. # # According to docs, this is equivalent to # `os.path.normpath(os.path.join(os.getcwd(), path))`. # The `os.getcwd()` call is slow here, so we precompute it. # # Additionally, since we are using normalized path already, # we skip calling `os.path.normpath` entirely. "abs_path": os.path.join(cwd, abs_path), "module": module, "filename": filename_for_module(module, abs_path) or None, "function": fid[2], "lineno": raw_frame.f_lineno, } if PY311: def get_frame_name(frame): # type: (FrameType) -> str return frame.f_code.co_qualname else: def get_frame_name(frame): # type: (FrameType) -> str f_code = frame.f_code co_varnames = f_code.co_varnames # co_name only contains the frame name. If the frame was a method, # the class name will NOT be included. name = f_code.co_name # if it was a method, we can get the class name by inspecting # the f_locals for the `self` argument try: if ( # the co_varnames start with the frame's positional arguments # and we expect the first to be `self` if its an instance method co_varnames and co_varnames[0] == "self" and "self" in frame.f_locals ): for cls in frame.f_locals["self"].__class__.__mro__: if name in cls.__dict__: return "{}.{}".format(cls.__name__, name) except AttributeError: pass # if it was a class method, (decorated with `@classmethod`) # we can get the class name by inspecting the f_locals for the `cls` argument try: if ( # the co_varnames start with the frame's positional arguments # and we expect the first to be `cls` if its a class method co_varnames and co_varnames[0] == "cls" and "cls" in frame.f_locals ): for cls in frame.f_locals["cls"].__mro__: if name in cls.__dict__: return "{}.{}".format(cls.__name__, name) except AttributeError: pass # nothing we can do if it is a staticmethod (decorated with @staticmethod) # we've done all we can, time to give up and return what we have return name MAX_PROFILE_DURATION_NS = int(3e10) # 30 seconds def get_current_thread_id(thread=None): # type: (Optional[threading.Thread]) -> Optional[int] """ Try to get the id of the current thread, with various fall backs. """ # if a thread is specified, that takes priority if thread is not None: try: thread_id = thread.ident if thread_id is not None: return thread_id except AttributeError: pass # if the app is using gevent, we should look at the gevent hub first # as the id there differs from what the threading module reports if is_gevent(): gevent_hub = get_gevent_hub() if gevent_hub is not None: try: # this is undocumented, so wrap it in try except to be safe return gevent_hub.thread_ident except AttributeError: pass # use the current thread's id if possible try: current_thread_id = threading.current_thread().ident if current_thread_id is not None: return current_thread_id except AttributeError: pass # if we can't get the current thread id, fall back to the main thread id try: main_thread_id = threading.main_thread().ident if main_thread_id is not None: return main_thread_id except AttributeError: pass # we've tried everything, time to give up return None class Profile(object): def __init__( self, transaction, # type: sentry_sdk.tracing.Transaction hub=None, # type: Optional[sentry_sdk.Hub] scheduler=None, # type: Optional[Scheduler] ): # type: (...) -> None self.scheduler = _scheduler if scheduler is None else scheduler self.hub = hub self.event_id = uuid.uuid4().hex # type: str # Here, we assume that the sampling decision on the transaction has been finalized. # # We cannot keep a reference to the transaction around here because it'll create # a reference cycle. So we opt to pull out just the necessary attributes. self.sampled = transaction.sampled # type: Optional[bool] # Various framework integrations are capable of overwriting the active thread id. # If it is set to `None` at the end of the profile, we fall back to the default. self._default_active_thread_id = get_current_thread_id() or 0 # type: int self.active_thread_id = None # type: Optional[int] try: self.start_ns = transaction._start_timestamp_monotonic_ns # type: int except AttributeError: self.start_ns = 0 self.stop_ns = 0 # type: int self.active = False # type: bool self.indexed_frames = {} # type: Dict[FrameId, int] self.indexed_stacks = {} # type: Dict[StackId, int] self.frames = [] # type: List[ProcessedFrame] self.stacks = [] # type: List[ProcessedStack] self.samples = [] # type: List[ProcessedSample] self.unique_samples = 0 transaction._profile = self def update_active_thread_id(self): # type: () -> None self.active_thread_id = get_current_thread_id() logger.debug( "[Profiling] updating active thread id to {tid}".format( tid=self.active_thread_id ) ) def _set_initial_sampling_decision(self, sampling_context): # type: (SamplingContext) -> None """ Sets the profile's sampling decision according to the following precdence rules: 1. If the transaction to be profiled is not sampled, that decision will be used, regardless of anything else. 2. Use `profiles_sample_rate` to decide. """ # The corresponding transaction was not sampled, # so don't generate a profile for it. if not self.sampled: logger.debug( "[Profiling] Discarding profile because transaction is discarded." ) self.sampled = False return # The profiler hasn't been properly initialized. if self.scheduler is None: logger.debug( "[Profiling] Discarding profile because profiler was not started." ) self.sampled = False return hub = self.hub or sentry_sdk.Hub.current client = hub.client # The client is None, so we can't get the sample rate. if client is None: self.sampled = False return options = client.options if callable(options.get("profiles_sampler")): sample_rate = options["profiles_sampler"](sampling_context) elif options["profiles_sample_rate"] is not None: sample_rate = options["profiles_sample_rate"] else: sample_rate = options["_experiments"].get("profiles_sample_rate") # The profiles_sample_rate option was not set, so profiling # was never enabled. if sample_rate is None: logger.debug( "[Profiling] Discarding profile because profiling was not enabled." ) self.sampled = False return if not is_valid_sample_rate(sample_rate, source="Profiling"): logger.warning( "[Profiling] Discarding profile because of invalid sample rate." ) self.sampled = False return # Now we roll the dice. random.random is inclusive of 0, but not of 1, # so strict < is safe here. In case sample_rate is a boolean, cast it # to a float (True becomes 1.0 and False becomes 0.0) self.sampled = random.random() < float(sample_rate) if self.sampled: logger.debug("[Profiling] Initializing profile") else: logger.debug( "[Profiling] Discarding profile because it's not included in the random sample (sample rate = {sample_rate})".format( sample_rate=float(sample_rate) ) ) def start(self): # type: () -> None if not self.sampled or self.active: return assert self.scheduler, "No scheduler specified" logger.debug("[Profiling] Starting profile") self.active = True if not self.start_ns: self.start_ns = nanosecond_time() self.scheduler.start_profiling(self) def stop(self): # type: () -> None if not self.sampled or not self.active: return assert self.scheduler, "No scheduler specified" logger.debug("[Profiling] Stopping profile") self.active = False self.scheduler.stop_profiling(self) self.stop_ns = nanosecond_time() def __enter__(self): # type: () -> Profile hub = self.hub or sentry_sdk.Hub.current _, scope = hub._stack[-1] old_profile = scope.profile scope.profile = self self._context_manager_state = (hub, scope, old_profile) self.start() return self def __exit__(self, ty, value, tb): # type: (Optional[Any], Optional[Any], Optional[Any]) -> None self.stop() _, scope, old_profile = self._context_manager_state del self._context_manager_state scope.profile = old_profile def write(self, ts, sample): # type: (int, ExtractedSample) -> None if not self.active: return if ts < self.start_ns: return offset = ts - self.start_ns if offset > MAX_PROFILE_DURATION_NS: self.stop() return self.unique_samples += 1 elapsed_since_start_ns = str(offset) for tid, (stack_id, frame_ids, frames) in sample: try: # Check if the stack is indexed first, this lets us skip # indexing frames if it's not necessary if stack_id not in self.indexed_stacks: for i, frame_id in enumerate(frame_ids): if frame_id not in self.indexed_frames: self.indexed_frames[frame_id] = len(self.indexed_frames) self.frames.append(frames[i]) self.indexed_stacks[stack_id] = len(self.indexed_stacks) self.stacks.append( [self.indexed_frames[frame_id] for frame_id in frame_ids] ) self.samples.append( { "elapsed_since_start_ns": elapsed_since_start_ns, "thread_id": tid, "stack_id": self.indexed_stacks[stack_id], } ) except AttributeError: # For some reason, the frame we get doesn't have certain attributes. # When this happens, we abandon the current sample as it's bad. capture_internal_exception(sys.exc_info()) def process(self): # type: () -> ProcessedProfile # This collects the thread metadata at the end of a profile. Doing it # this way means that any threads that terminate before the profile ends # will not have any metadata associated with it. thread_metadata = { str(thread.ident): { "name": str(thread.name), } for thread in threading.enumerate() } # type: Dict[str, ProcessedThreadMetadata] return { "frames": self.frames, "stacks": self.stacks, "samples": self.samples, "thread_metadata": thread_metadata, } def to_json(self, event_opt, options): # type: (Any, Dict[str, Any], Dict[str, Any]) -> Dict[str, Any] profile = self.process() set_in_app_in_frames( profile["frames"], options["in_app_exclude"], options["in_app_include"], options["project_root"], ) return { "environment": event_opt.get("environment"), "event_id": self.event_id, "platform": "python", "profile": profile, "release": event_opt.get("release", ""), "timestamp": event_opt["start_timestamp"], "version": "1", "device": { "architecture": platform.machine(), }, "os": { "name": platform.system(), "version": platform.release(), }, "runtime": { "name": platform.python_implementation(), "version": platform.python_version(), }, "transactions": [ { "id": event_opt["event_id"], "name": event_opt["transaction"], # we start the transaction before the profile and this is # the transaction start time relative to the profile, so we # hardcode it to 0 until we can start the profile before "relative_start_ns": "0", # use the duration of the profile instead of the transaction # because we end the transaction after the profile "relative_end_ns": str(self.stop_ns - self.start_ns), "trace_id": event_opt["contexts"]["trace"]["trace_id"], "active_thread_id": str( self._default_active_thread_id if self.active_thread_id is None else self.active_thread_id ), } ], } def valid(self): # type: () -> bool hub = self.hub or sentry_sdk.Hub.current client = hub.client if client is None: return False if not has_profiling_enabled(client.options): return False if self.sampled is None or not self.sampled: if client.transport: client.transport.record_lost_event( "sample_rate", data_category="profile" ) return False if self.unique_samples < PROFILE_MINIMUM_SAMPLES: if client.transport: client.transport.record_lost_event( "insufficient_data", data_category="profile" ) logger.debug("[Profiling] Discarding profile because insufficient samples.") return False return True class Scheduler(object): mode = "unknown" # type: ProfilerMode def __init__(self, frequency): # type: (int) -> None self.interval = 1.0 / frequency self.sampler = self.make_sampler() # cap the number of new profiles at any time so it does not grow infinitely self.new_profiles = deque(maxlen=128) # type: Deque[Profile] self.active_profiles = set() # type: Set[Profile] def __enter__(self): # type: () -> Scheduler self.setup() return self def __exit__(self, ty, value, tb): # type: (Optional[Any], Optional[Any], Optional[Any]) -> None self.teardown() def setup(self): # type: () -> None raise NotImplementedError def teardown(self): # type: () -> None raise NotImplementedError def ensure_running(self): # type: () -> None raise NotImplementedError def start_profiling(self, profile): # type: (Profile) -> None self.ensure_running() self.new_profiles.append(profile) def stop_profiling(self, profile): # type: (Profile) -> None pass def make_sampler(self): # type: () -> Callable[..., None] cwd = os.getcwd() cache = LRUCache(max_size=256) def _sample_stack(args, *kwargs): # type: (Any, **Any) -> None """ Take a sample of the stack on all the threads in the process. This should be called at a regular interval to collect samples. """ # no profiles taking place, so we can stop early if not self.new_profiles and not self.active_profiles: # make sure to clear the cache if we're not profiling so we dont # keep a reference to the last stack of frames around return # This is the number of profiles we want to pop off. # It's possible another thread adds a new profile to # the list and we spend longer than we want inside # the loop below. # # Also make sure to set this value before extracting # frames so we do not write to any new profiles that # were started after this point. new_profiles = len(self.new_profiles) now = nanosecond_time() try: sample = [ (str(tid), extract_stack(frame, cache, cwd)) for tid, frame in sys._current_frames().items() ] except AttributeError: # For some reason, the frame we get doesn't have certain attributes. # When this happens, we abandon the current sample as it's bad. capture_internal_exception(sys.exc_info()) return # Move the new profiles into the active_profiles set. # # We cannot directly add the to active_profiles set # in `start_profiling` because it is called from other # threads which can cause a RuntimeError when it the # set sizes changes during iteration without a lock. # # We also want to avoid using a lock here so threads # that are starting profiles are not blocked until it # can acquire the lock. for _ in range(new_profiles): self.active_profiles.add(self.new_profiles.popleft()) inactive_profiles = [] for profile in self.active_profiles: if profile.active: profile.write(now, sample) else: # If a thread is marked inactive, we buffer it # to `inactive_profiles` so it can be removed. # We cannot remove it here as it would result # in a RuntimeError. inactive_profiles.append(profile) for profile in inactive_profiles: self.active_profiles.remove(profile) return _sample_stack class ThreadScheduler(Scheduler): """ This scheduler is based on running a daemon thread that will call the sampler at a regular interval. """ mode = "thread" # type: ProfilerMode name = "sentry.profiler.ThreadScheduler" def __init__(self, frequency): # type: (int) -> None super(ThreadScheduler, self).__init__(frequency=frequency) # used to signal to the thread that it should stop self.running = False self.thread = None # type: Optional[threading.Thread] self.pid = None # type: Optional[int] self.lock = threading.Lock() def setup(self): # type: () -> None pass def teardown(self): # type: () -> None if self.running: self.running = False if self.thread is not None: self.thread.join() def ensure_running(self): # type: () -> None pid = os.getpid() # is running on the right process if self.running and self.pid == pid: return with self.lock: # another thread may have tried to acquire the lock # at the same time so it may start another thread # make sure to check again before proceeding if self.running and self.pid == pid: return self.pid = pid self.running = True # make sure the thread is a daemon here otherwise this # can keep the application running after other threads # have exited self.thread = threading.Thread(name=self.name, target=self.run, daemon=True) self.thread.start() def run(self): # type: () -> None last = time.perf_counter() while self.running: self.sampler() # some time may have elapsed since the last time # we sampled, so we need to account for that and # not sleep for too long elapsed = time.perf_counter() - last if elapsed < self.interval: thread_sleep(self.interval - elapsed) # after sleeping, make sure to take the current # timestamp so we can use it next iteration last = time.perf_counter() class GeventScheduler(Scheduler): """ This scheduler is based on the thread scheduler but adapted to work with gevent. When using gevent, it may monkey patch the threading modules (`threading` and `_thread`). This results in the use of greenlets instead of native threads. This is an issue because the sampler CANNOT run in a greenlet because 1. Other greenlets doing sync work will prevent the sampler from running 2. The greenlet runs in the same thread as other greenlets so when taking a sample, other greenlets will have been evicted from the thread. This results in a sample containing only the sampler's code. """ mode = "gevent" # type: ProfilerMode name = "sentry.profiler.GeventScheduler" def __init__(self, frequency): # type: (int) -> None if ThreadPool is None: raise ValueError("Profiler mode: {} is not available".format(self.mode)) super(GeventScheduler, self).__init__(frequency=frequency) # used to signal to the thread that it should stop self.running = False self.thread = None # type: Optional[ThreadPool] self.pid = None # type: Optional[int] # This intentionally uses the gevent patched threading.Lock. # The lock will be required when first trying to start profiles # as we need to spawn the profiler thread from the greenlets. self.lock = threading.Lock() def setup(self): # type: () -> None pass def teardown(self): # type: () -> None if self.running: self.running = False if self.thread is not None: self.thread.join() def ensure_running(self): # type: () -> None pid = os.getpid() # is running on the right process if self.running and self.pid == pid: return with self.lock: # another thread may have tried to acquire the lock # at the same time so it may start another thread # make sure to check again before proceeding if self.running and self.pid == pid: return self.pid = pid self.running = True self.thread = ThreadPool(1) self.thread.spawn(self.run) def run(self): # type: () -> None last = time.perf_counter() while self.running: self.sampler() # some time may have elapsed since the last time # we sampled, so we need to account for that and # not sleep for too long elapsed = time.perf_counter() - last if elapsed < self.interval: thread_sleep(self.interval - elapsed) # after sleeping, make sure to take the current # timestamp so we can use it next iteration last = time.perf_counter()	33.18 KB	2026-01-20 12:58
📄 py.typed	0 B	2026-01-20 12:58
📄 scope.py from copy import copy from collections import deque from itertools import chain import os import uuid from sentry_sdk.attachments import Attachment from sentry_sdk._functools import wraps from sentry_sdk.tracing_utils import ( Baggage, extract_sentrytrace_data, has_tracing_enabled, normalize_incoming_data, ) from sentry_sdk.tracing import ( BAGGAGE_HEADER_NAME, SENTRY_TRACE_HEADER_NAME, Transaction, ) from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.utils import logger, capture_internal_exceptions from sentry_sdk.consts import FALSE_VALUES if TYPE_CHECKING: from typing import Any from typing import Dict from typing import Iterator from typing import Optional from typing import Deque from typing import List from typing import Callable from typing import Tuple from typing import TypeVar from sentry_sdk._types import ( Breadcrumb, Event, EventProcessor, ErrorProcessor, ExcInfo, Hint, Type, ) from sentry_sdk.profiler import Profile from sentry_sdk.tracing import Span from sentry_sdk.session import Session F = TypeVar("F", bound=Callable[..., Any]) T = TypeVar("T") global_event_processors = [] # type: List[EventProcessor] def add_global_event_processor(processor): # type: (EventProcessor) -> None global_event_processors.append(processor) def _attr_setter(fn): # type: (Any) -> Any return property(fset=fn, doc=fn.__doc__) def _disable_capture(fn): # type: (F) -> F @wraps(fn) def wrapper(self, args, kwargs): # type: (Any, Dict[str, Any], *Any) -> Any if not self._should_capture: return try: self._should_capture = False return fn(self, args, *kwargs) finally: self._should_capture = True return wrapper # type: ignore class Scope(object): """The scope holds extra information that should be sent with all events that belong to it. """ # NOTE: Even though it should not happen, the scope needs to not crash when # accessed by multiple threads. It's fine if it's full of races, but those # races should never make the user application crash. # # The same needs to hold for any accesses of the scope the SDK makes. __slots__ = ( "_level", "_name", "_fingerprint", # note that for legacy reasons, _transaction is the transaction name*, # not a Transaction object (the object is stored in _span) "_transaction", "_transaction_info", "_user", "_tags", "_contexts", "_extras", "_breadcrumbs", "_event_processors", "_error_processors", "_should_capture", "_span", "_session", "_attachments", "_force_auto_session_tracking", "_profile", "_propagation_context", ) def __init__(self): # type: () -> None self._event_processors = [] # type: List[EventProcessor] self._error_processors = [] # type: List[ErrorProcessor] self._name = None # type: Optional[str] self._propagation_context = None # type: Optional[Dict[str, Any]] self.clear() incoming_trace_information = self._load_trace_data_from_env() self.generate_propagation_context(incoming_data=incoming_trace_information) def _load_trace_data_from_env(self): # type: () -> Optional[Dict[str, str]] """ Load Sentry trace id and baggage from environment variables. Can be disabled by setting SENTRY_USE_ENVIRONMENT to "false". """ incoming_trace_information = None sentry_use_environment = ( os.environ.get("SENTRY_USE_ENVIRONMENT") or "" ).lower() use_environment = sentry_use_environment not in FALSE_VALUES if use_environment: incoming_trace_information = {} if os.environ.get("SENTRY_TRACE"): incoming_trace_information[SENTRY_TRACE_HEADER_NAME] = ( os.environ.get("SENTRY_TRACE") or "" ) if os.environ.get("SENTRY_BAGGAGE"): incoming_trace_information[BAGGAGE_HEADER_NAME] = ( os.environ.get("SENTRY_BAGGAGE") or "" ) return incoming_trace_information or None def _extract_propagation_context(self, data): # type: (Dict[str, Any]) -> Optional[Dict[str, Any]] context = {} # type: Dict[str, Any] normalized_data = normalize_incoming_data(data) baggage_header = normalized_data.get(BAGGAGE_HEADER_NAME) if baggage_header: context["dynamic_sampling_context"] = Baggage.from_incoming_header( baggage_header ).dynamic_sampling_context() sentry_trace_header = normalized_data.get(SENTRY_TRACE_HEADER_NAME) if sentry_trace_header: sentrytrace_data = extract_sentrytrace_data(sentry_trace_header) if sentrytrace_data is not None: context.update(sentrytrace_data) only_baggage_no_sentry_trace = ( "dynamic_sampling_context" in context and "trace_id" not in context ) if only_baggage_no_sentry_trace: context.update(self._create_new_propagation_context()) if context: if not context.get("span_id"): context["span_id"] = uuid.uuid4().hex[16:] return context return None def _create_new_propagation_context(self): # type: () -> Dict[str, Any] return { "trace_id": uuid.uuid4().hex, "span_id": uuid.uuid4().hex[16:], "parent_span_id": None, "dynamic_sampling_context": None, } def set_new_propagation_context(self): # type: () -> None """ Creates a new propagation context and sets it as `_propagation_context`. Overwriting existing one. """ self._propagation_context = self._create_new_propagation_context() logger.debug( "[Tracing] Create new propagation context: %s", self._propagation_context, ) def generate_propagation_context(self, incoming_data=None): # type: (Optional[Dict[str, str]]) -> None """ Makes sure `_propagation_context` is set. If there is `incoming_data` overwrite existing `_propagation_context`. if there is no `incoming_data` create new `_propagation_context`, but do NOT overwrite if already existing. """ if incoming_data: context = self._extract_propagation_context(incoming_data) if context is not None: self._propagation_context = context logger.debug( "[Tracing] Extracted propagation context from incoming data: %s", self._propagation_context, ) if self._propagation_context is None: self.set_new_propagation_context() def get_dynamic_sampling_context(self): # type: () -> Optional[Dict[str, str]] """ Returns the Dynamic Sampling Context from the Propagation Context. If not existing, creates a new one. """ if self._propagation_context is None: return None baggage = self.get_baggage() if baggage is not None: self._propagation_context[ "dynamic_sampling_context" ] = baggage.dynamic_sampling_context() return self._propagation_context["dynamic_sampling_context"] def get_traceparent(self): # type: () -> Optional[str] """ Returns the Sentry "sentry-trace" header (aka the traceparent) from the Propagation Context. """ if self._propagation_context is None: return None traceparent = "%s-%s" % ( self._propagation_context["trace_id"], self._propagation_context["span_id"], ) return traceparent def get_baggage(self): # type: () -> Optional[Baggage] if self._propagation_context is None: return None dynamic_sampling_context = self._propagation_context.get( "dynamic_sampling_context" ) if dynamic_sampling_context is None: return Baggage.from_options(self) else: return Baggage(dynamic_sampling_context) def get_trace_context(self): # type: () -> Any """ Returns the Sentry "trace" context from the Propagation Context. """ if self._propagation_context is None: return None trace_context = { "trace_id": self._propagation_context["trace_id"], "span_id": self._propagation_context["span_id"], "parent_span_id": self._propagation_context["parent_span_id"], "dynamic_sampling_context": self.get_dynamic_sampling_context(), } # type: Dict[str, Any] return trace_context def iter_headers(self): # type: () -> Iterator[Tuple[str, str]] """ Creates a generator which returns the `sentry-trace` and `baggage` headers from the Propagation Context. """ if self._propagation_context is not None: traceparent = self.get_traceparent() if traceparent is not None: yield SENTRY_TRACE_HEADER_NAME, traceparent dsc = self.get_dynamic_sampling_context() if dsc is not None: baggage = Baggage(dsc).serialize() yield BAGGAGE_HEADER_NAME, baggage def clear(self): # type: () -> None """Clears the entire scope.""" self._level = None # type: Optional[str] self._fingerprint = None # type: Optional[List[str]] self._transaction = None # type: Optional[str] self._transaction_info = {} # type: Dict[str, str] self._user = None # type: Optional[Dict[str, Any]] self._tags = {} # type: Dict[str, Any] self._contexts = {} # type: Dict[str, Dict[str, Any]] self._extras = {} # type: Dict[str, Any] self._attachments = [] # type: List[Attachment] self.clear_breadcrumbs() self._should_capture = True self._span = None # type: Optional[Span] self._session = None # type: Optional[Session] self._force_auto_session_tracking = None # type: Optional[bool] self._profile = None # type: Optional[Profile] self._propagation_context = None @_attr_setter def level(self, value): # type: (Optional[str]) -> None """When set this overrides the level. Deprecated in favor of set_level.""" self._level = value def set_level(self, value): # type: (Optional[str]) -> None """Sets the level for the scope.""" self._level = value @_attr_setter def fingerprint(self, value): # type: (Optional[List[str]]) -> None """When set this overrides the default fingerprint.""" self._fingerprint = value @property def transaction(self): # type: () -> Any # would be type: () -> Optional[Transaction], see https://github.com/python/mypy/issues/3004 """Return the transaction (root span) in the scope, if any.""" # there is no span/transaction on the scope if self._span is None: return None # there is an orphan span on the scope if self._span.containing_transaction is None: return None # there is either a transaction (which is its own containing # transaction) or a non-orphan span on the scope return self._span.containing_transaction @transaction.setter def transaction(self, value): # type: (Any) -> None # would be type: (Optional[str]) -> None, see https://github.com/python/mypy/issues/3004 """When set this forces a specific transaction name to be set. Deprecated: use set_transaction_name instead.""" # XXX: the docstring above is misleading. The implementation of # apply_to_event prefers an existing value of event.transaction over # anything set in the scope. # XXX: note that with the introduction of the Scope.transaction getter, # there is a semantic and type mismatch between getter and setter. The # getter returns a Transaction, the setter sets a transaction name. # Without breaking version compatibility, we could make the setter set a # transaction name or transaction (self._span) depending on the type of # the value argument. logger.warning( "Assigning to scope.transaction directly is deprecated: use scope.set_transaction_name() instead." ) self._transaction = value if self._span and self._span.containing_transaction: self._span.containing_transaction.name = value def set_transaction_name(self, name, source=None): # type: (str, Optional[str]) -> None """Set the transaction name and optionally the transaction source.""" self._transaction = name if self._span and self._span.containing_transaction: self._span.containing_transaction.name = name if source: self._span.containing_transaction.source = source if source: self._transaction_info["source"] = source @_attr_setter def user(self, value): # type: (Optional[Dict[str, Any]]) -> None """When set a specific user is bound to the scope. Deprecated in favor of set_user.""" self.set_user(value) def set_user(self, value): # type: (Optional[Dict[str, Any]]) -> None """Sets a user for the scope.""" self._user = value if self._session is not None: self._session.update(user=value) @property def span(self): # type: () -> Optional[Span] """Get/set current tracing span or transaction.""" return self._span @span.setter def span(self, span): # type: (Optional[Span]) -> None self._span = span # XXX: this differs from the implementation in JS, there Scope.setSpan # does not set Scope._transactionName. if isinstance(span, Transaction): transaction = span if transaction.name: self._transaction = transaction.name @property def profile(self): # type: () -> Optional[Profile] return self._profile @profile.setter def profile(self, profile): # type: (Optional[Profile]) -> None self._profile = profile def set_tag( self, key, # type: str value, # type: Any ): # type: (...) -> None """Sets a tag for a key to a specific value.""" self._tags[key] = value def remove_tag( self, key # type: str ): # type: (...) -> None """Removes a specific tag.""" self._tags.pop(key, None) def set_context( self, key, # type: str value, # type: Dict[str, Any] ): # type: (...) -> None """Binds a context at a certain key to a specific value.""" self._contexts[key] = value def remove_context( self, key # type: str ): # type: (...) -> None """Removes a context.""" self._contexts.pop(key, None) def set_extra( self, key, # type: str value, # type: Any ): # type: (...) -> None """Sets an extra key to a specific value.""" self._extras[key] = value def remove_extra( self, key # type: str ): # type: (...) -> None """Removes a specific extra key.""" self._extras.pop(key, None) def clear_breadcrumbs(self): # type: () -> None """Clears breadcrumb buffer.""" self._breadcrumbs = deque() # type: Deque[Breadcrumb] def add_attachment( self, bytes=None, # type: Optional[bytes] filename=None, # type: Optional[str] path=None, # type: Optional[str] content_type=None, # type: Optional[str] add_to_transactions=False, # type: bool ): # type: (...) -> None """Adds an attachment to future events sent.""" self._attachments.append( Attachment( bytes=bytes, path=path, filename=filename, content_type=content_type, add_to_transactions=add_to_transactions, ) ) def add_event_processor( self, func # type: EventProcessor ): # type: (...) -> None """Register a scope local event processor on the scope. :param func: This function behaves like `before_send.` """ if len(self._event_processors) > 20: logger.warning( "Too many event processors on scope! Clearing list to free up some memory: %r", self._event_processors, ) del self._event_processors[:] self._event_processors.append(func) def add_error_processor( self, func, # type: ErrorProcessor cls=None, # type: Optional[Type[BaseException]] ): # type: (...) -> None """Register a scope local error processor on the scope. :param func: A callback that works similar to an event processor but is invoked with the original exception info triple as second argument. :param cls: Optionally, only process exceptions of this type. """ if cls is not None: cls_ = cls # For mypy. real_func = func def func(event, exc_info): # type: (Event, ExcInfo) -> Optional[Event] try: is_inst = isinstance(exc_info[1], cls_) except Exception: is_inst = False if is_inst: return real_func(event, exc_info) return event self._error_processors.append(func) @_disable_capture def apply_to_event( self, event, # type: Event hint, # type: Hint options=None, # type: Optional[Dict[str, Any]] ): # type: (...) -> Optional[Event] """Applies the information contained on the scope to the given event.""" def _drop(cause, ty): # type: (Any, str) -> Optional[Any] logger.info("%s (%s) dropped event", ty, cause) return None is_transaction = event.get("type") == "transaction" # put all attachments into the hint. This lets callbacks play around # with attachments. We also later pull this out of the hint when we # create the envelope. attachments_to_send = hint.get("attachments") or [] for attachment in self._attachments: if not is_transaction or attachment.add_to_transactions: attachments_to_send.append(attachment) hint["attachments"] = attachments_to_send if self._level is not None: event["level"] = self._level if not is_transaction: event.setdefault("breadcrumbs", {}).setdefault("values", []).extend( self._breadcrumbs ) if event.get("user") is None and self._user is not None: event["user"] = self._user if event.get("transaction") is None and self._transaction is not None: event["transaction"] = self._transaction if event.get("transaction_info") is None and self._transaction_info is not None: event["transaction_info"] = self._transaction_info if event.get("fingerprint") is None and self._fingerprint is not None: event["fingerprint"] = self._fingerprint if self._extras: event.setdefault("extra", {}).update(self._extras) if self._tags: event.setdefault("tags", {}).update(self._tags) if self._contexts: event.setdefault("contexts", {}).update(self._contexts) contexts = event.setdefault("contexts", {}) if contexts.get("trace") is None: if has_tracing_enabled(options) and self._span is not None: contexts["trace"] = self._span.get_trace_context() else: contexts["trace"] = self.get_trace_context() try: replay_id = contexts["trace"]["dynamic_sampling_context"]["replay_id"] except (KeyError, TypeError): replay_id = None if replay_id is not None: contexts["replay"] = { "replay_id": replay_id, } exc_info = hint.get("exc_info") if exc_info is not None: for error_processor in self._error_processors: new_event = error_processor(event, exc_info) if new_event is None: return _drop(error_processor, "error processor") event = new_event for event_processor in chain(global_event_processors, self._event_processors): new_event = event with capture_internal_exceptions(): new_event = event_processor(event, hint) if new_event is None: return _drop(event_processor, "event processor") event = new_event return event def update_from_scope(self, scope): # type: (Scope) -> None if scope._level is not None: self._level = scope._level if scope._fingerprint is not None: self._fingerprint = scope._fingerprint if scope._transaction is not None: self._transaction = scope._transaction if scope._transaction_info is not None: self._transaction_info.update(scope._transaction_info) if scope._user is not None: self._user = scope._user if scope._tags: self._tags.update(scope._tags) if scope._contexts: self._contexts.update(scope._contexts) if scope._extras: self._extras.update(scope._extras) if scope._breadcrumbs: self._breadcrumbs.extend(scope._breadcrumbs) if scope._span: self._span = scope._span if scope._attachments: self._attachments.extend(scope._attachments) if scope._profile: self._profile = scope._profile if scope._propagation_context: self._propagation_context = scope._propagation_context def update_from_kwargs( self, user=None, # type: Optional[Any] level=None, # type: Optional[str] extras=None, # type: Optional[Dict[str, Any]] contexts=None, # type: Optional[Dict[str, Any]] tags=None, # type: Optional[Dict[str, str]] fingerprint=None, # type: Optional[List[str]] ): # type: (...) -> None if level is not None: self._level = level if user is not None: self._user = user if extras is not None: self._extras.update(extras) if contexts is not None: self._contexts.update(contexts) if tags is not None: self._tags.update(tags) if fingerprint is not None: self._fingerprint = fingerprint def __copy__(self): # type: () -> Scope rv = object.__new__(self.__class__) # type: Scope rv._level = self._level rv._name = self._name rv._fingerprint = self._fingerprint rv._transaction = self._transaction rv._transaction_info = dict(self._transaction_info) rv._user = self._user rv._tags = dict(self._tags) rv._contexts = dict(self._contexts) rv._extras = dict(self._extras) rv._breadcrumbs = copy(self._breadcrumbs) rv._event_processors = list(self._event_processors) rv._error_processors = list(self._error_processors) rv._propagation_context = self._propagation_context rv._should_capture = self._should_capture rv._span = self._span rv._session = self._session rv._force_auto_session_tracking = self._force_auto_session_tracking rv._attachments = list(self._attachments) rv._profile = self._profile return rv def __repr__(self): # type: () -> str return "<%s id=%s name=%s>" % ( self.__class__.__name__, hex(id(self)), self._name, )	24.26 KB	2026-01-20 12:58
📄 scrubber.py from sentry_sdk.utils import ( capture_internal_exceptions, AnnotatedValue, iter_event_frames, ) from sentry_sdk._compat import string_types from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from sentry_sdk._types import Event from typing import Any from typing import Dict from typing import List from typing import Optional DEFAULT_DENYLIST = [ # stolen from relay "password", "passwd", "secret", "api_key", "apikey", "auth", "credentials", "mysql_pwd", "privatekey", "private_key", "token", "ip_address", "session", # django "csrftoken", "sessionid", # wsgi "remote_addr", "x_csrftoken", "x_forwarded_for", "set_cookie", "cookie", "authorization", "x_api_key", "x_forwarded_for", "x_real_ip", # other common names used in the wild "aiohttp_session", # aiohttp "connect.sid", # Express "csrf_token", # Pyramid "csrf", # (this is a cookie name used in accepted answers on stack overflow) "_csrf", # Express "_csrf_token", # Bottle "PHPSESSID", # PHP "_session", # Sanic "symfony", # Symfony "user_session", # Vue "_xsrf", # Tornado "XSRF-TOKEN", # Angular, Laravel ] class EventScrubber(object): def __init__(self, denylist=None): # type: (Optional[List[str]]) -> None self.denylist = DEFAULT_DENYLIST if denylist is None else denylist self.denylist = [x.lower() for x in self.denylist] def scrub_dict(self, d): # type: (Dict[str, Any]) -> None if not isinstance(d, dict): return for k in d.keys(): if isinstance(k, string_types) and k.lower() in self.denylist: d[k] = AnnotatedValue.substituted_because_contains_sensitive_data() def scrub_request(self, event): # type: (Event) -> None with capture_internal_exceptions(): if "request" in event: if "headers" in event["request"]: self.scrub_dict(event["request"]["headers"]) if "cookies" in event["request"]: self.scrub_dict(event["request"]["cookies"]) if "data" in event["request"]: self.scrub_dict(event["request"]["data"]) def scrub_extra(self, event): # type: (Event) -> None with capture_internal_exceptions(): if "extra" in event: self.scrub_dict(event["extra"]) def scrub_user(self, event): # type: (Event) -> None with capture_internal_exceptions(): if "user" in event: self.scrub_dict(event["user"]) def scrub_breadcrumbs(self, event): # type: (Event) -> None with capture_internal_exceptions(): if "breadcrumbs" in event: if "values" in event["breadcrumbs"]: for value in event["breadcrumbs"]["values"]: if "data" in value: self.scrub_dict(value["data"]) def scrub_frames(self, event): # type: (Event) -> None with capture_internal_exceptions(): for frame in iter_event_frames(event): if "vars" in frame: self.scrub_dict(frame["vars"]) def scrub_spans(self, event): # type: (Event) -> None with capture_internal_exceptions(): if "spans" in event: for span in event["spans"]: if "data" in span: self.scrub_dict(span["data"]) def scrub_event(self, event): # type: (Event) -> None self.scrub_request(event) self.scrub_extra(event) self.scrub_user(event) self.scrub_breadcrumbs(event) self.scrub_frames(event) self.scrub_spans(event)	3.8 KB	2026-01-20 12:58
📄 serializer.py import sys import math from datetime import datetime from sentry_sdk.utils import ( AnnotatedValue, capture_internal_exception, disable_capture_event, format_timestamp, safe_repr, strip_string, ) from sentry_sdk._compat import ( text_type, PY2, string_types, number_types, iteritems, binary_sequence_types, ) from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from types import TracebackType from typing import Any from typing import Callable from typing import ContextManager from typing import Dict from typing import List from typing import Optional from typing import Type from typing import Union from sentry_sdk._types import NotImplementedType, Event Span = Dict[str, Any] ReprProcessor = Callable[[Any, Dict[str, Any]], Union[NotImplementedType, str]] Segment = Union[str, int] if PY2: # Importing ABCs from collections is deprecated, and will stop working in 3.8 # https://github.com/python/cpython/blob/master/Lib/collections/__init__.py#L49 from collections import Mapping, Sequence, Set serializable_str_types = string_types + binary_sequence_types else: # New in 3.3 # https://docs.python.org/3/library/collections.abc.html from collections.abc import Mapping, Sequence, Set # Bytes are technically not strings in Python 3, but we can serialize them serializable_str_types = string_types + binary_sequence_types # Maximum length of JSON-serialized event payloads that can be safely sent # before the server may reject the event due to its size. This is not intended # to reflect actual values defined server-side, but rather only be an upper # bound for events sent by the SDK. # # Can be overwritten if wanting to send more bytes, e.g. with a custom server. # When changing this, keep in mind that events may be a little bit larger than # this value due to attached metadata, so keep the number conservative. MAX_EVENT_BYTES = 106 # Maximum depth and breadth of databags. Excess data will be trimmed. If # max_request_body_size is "always", request bodies won't be trimmed. MAX_DATABAG_DEPTH = 5 MAX_DATABAG_BREADTH = 10 CYCLE_MARKER = "<cyclic>" global_repr_processors = [] # type: List[ReprProcessor] def add_global_repr_processor(processor): # type: (ReprProcessor) -> None global_repr_processors.append(processor) class Memo(object): __slots__ = ("_ids", "_objs") def __init__(self): # type: () -> None self._ids = {} # type: Dict[int, Any] self._objs = [] # type: List[Any] def memoize(self, obj): # type: (Any) -> ContextManager[bool] self._objs.append(obj) return self def __enter__(self): # type: () -> bool obj = self._objs[-1] if id(obj) in self._ids: return True else: self._ids[id(obj)] = obj return False def __exit__( self, ty, # type: Optional[Type[BaseException]] value, # type: Optional[BaseException] tb, # type: Optional[TracebackType] ): # type: (...) -> None self._ids.pop(id(self._objs.pop()), None) def serialize(event, kwargs): # type: (Event, Any) -> Event memo = Memo() path = [] # type: List[Segment] meta_stack = [] # type: List[Dict[str, Any]] keep_request_bodies = ( kwargs.pop("max_request_body_size", None) == "always" ) # type: bool max_value_length = kwargs.pop("max_value_length", None) # type: Optional[int] def _annotate(meta): # type: (Any) -> None while len(meta_stack) <= len(path): try: segment = path[len(meta_stack) - 1] node = meta_stack[-1].setdefault(text_type(segment), {}) except IndexError: node = {} meta_stack.append(node) meta_stack[-1].setdefault("", {}).update(meta) def _should_repr_strings(): # type: () -> Optional[bool] """ By default non-serializable objects are going through safe_repr(). For certain places in the event (local vars) we want to repr() even things that are JSON-serializable to make their type more apparent. For example, it's useful to see the difference between a unicode-string and a bytestring when viewing a stacktrace. For container-types we still don't do anything different. Generally we just try to make the Sentry UI present exactly what a pretty-printed repr would look like. :returns: `True` if we are somewhere in frame variables, and `False` if we are in a position where we will never encounter frame variables when recursing (for example, we're in `event.extra`). `None` if we are not (yet) in frame variables, but might encounter them when recursing (e.g. we're in `event.exception`) """ try: p0 = path[0] if p0 == "stacktrace" and path[1] == "frames" and path[3] == "vars": return True if ( p0 in ("threads", "exception") and path[1] == "values" and path[3] == "stacktrace" and path[4] == "frames" and path[6] == "vars" ): return True except IndexError: return None return False def _is_databag(): # type: () -> Optional[bool] """ A databag is any value that we need to trim. :returns: Works like `_should_repr_strings()`. `True` for "yes", `False` for :"no", `None` for "maybe soon". """ try: rv = _should_repr_strings() if rv in (True, None): return rv is_request_body = _is_request_body() if is_request_body in (True, None): return is_request_body p0 = path[0] if p0 == "breadcrumbs" and path[1] == "values": path[2] return True if p0 == "extra": return True except IndexError: return None return False def _is_request_body(): # type: () -> Optional[bool] try: if path[0] == "request" and path[1] == "data": return True except IndexError: return None return False def _serialize_node( obj, # type: Any is_databag=None, # type: Optional[bool] is_request_body=None, # type: Optional[bool] should_repr_strings=None, # type: Optional[bool] segment=None, # type: Optional[Segment] remaining_breadth=None, # type: Optional[Union[int, float]] remaining_depth=None, # type: Optional[Union[int, float]] ): # type: (...) -> Any if segment is not None: path.append(segment) try: with memo.memoize(obj) as result: if result: return CYCLE_MARKER return _serialize_node_impl( obj, is_databag=is_databag, is_request_body=is_request_body, should_repr_strings=should_repr_strings, remaining_depth=remaining_depth, remaining_breadth=remaining_breadth, ) except BaseException: capture_internal_exception(sys.exc_info()) if is_databag: return "<failed to serialize, use init(debug=True) to see error logs>" return None finally: if segment is not None: path.pop() del meta_stack[len(path) + 1 :] def _flatten_annotated(obj): # type: (Any) -> Any if isinstance(obj, AnnotatedValue): _annotate(obj.metadata) obj = obj.value return obj def _serialize_node_impl( obj, is_databag, is_request_body, should_repr_strings, remaining_depth, remaining_breadth, ): # type: (Any, Optional[bool], Optional[bool], Optional[bool], Optional[Union[float, int]], Optional[Union[float, int]]) -> Any if isinstance(obj, AnnotatedValue): should_repr_strings = False if should_repr_strings is None: should_repr_strings = _should_repr_strings() if is_databag is None: is_databag = _is_databag() if is_request_body is None: is_request_body = _is_request_body() if is_databag: if is_request_body and keep_request_bodies: remaining_depth = float("inf") remaining_breadth = float("inf") else: if remaining_depth is None: remaining_depth = MAX_DATABAG_DEPTH if remaining_breadth is None: remaining_breadth = MAX_DATABAG_BREADTH obj = _flatten_annotated(obj) if remaining_depth is not None and remaining_depth <= 0: _annotate(rem=[["!limit", "x"]]) if is_databag: return _flatten_annotated( strip_string(safe_repr(obj), max_length=max_value_length) ) return None if is_databag and global_repr_processors: hints = {"memo": memo, "remaining_depth": remaining_depth} for processor in global_repr_processors: result = processor(obj, hints) if result is not NotImplemented: return _flatten_annotated(result) sentry_repr = getattr(type(obj), "__sentry_repr__", None) if obj is None or isinstance(obj, (bool, number_types)): if should_repr_strings or ( isinstance(obj, float) and (math.isinf(obj) or math.isnan(obj)) ): return safe_repr(obj) else: return obj elif callable(sentry_repr): return sentry_repr(obj) elif isinstance(obj, datetime): return ( text_type(format_timestamp(obj)) if not should_repr_strings else safe_repr(obj) ) elif isinstance(obj, Mapping): # Create temporary copy here to avoid calling too much code that # might mutate our dictionary while we're still iterating over it. obj = dict(iteritems(obj)) rv_dict = {} # type: Dict[str, Any] i = 0 for k, v in iteritems(obj): if remaining_breadth is not None and i >= remaining_breadth: _annotate(len=len(obj)) break str_k = text_type(k) v = _serialize_node( v, segment=str_k, should_repr_strings=should_repr_strings, is_databag=is_databag, is_request_body=is_request_body, remaining_depth=remaining_depth - 1 if remaining_depth is not None else None, remaining_breadth=remaining_breadth, ) rv_dict[str_k] = v i += 1 return rv_dict elif not isinstance(obj, serializable_str_types) and isinstance( obj, (Set, Sequence) ): rv_list = [] for i, v in enumerate(obj): if remaining_breadth is not None and i >= remaining_breadth: _annotate(len=len(obj)) break rv_list.append( _serialize_node( v, segment=i, should_repr_strings=should_repr_strings, is_databag=is_databag, is_request_body=is_request_body, remaining_depth=remaining_depth - 1 if remaining_depth is not None else None, remaining_breadth=remaining_breadth, ) ) return rv_list if should_repr_strings: obj = safe_repr(obj) else: if isinstance(obj, bytes) or isinstance(obj, bytearray): obj = obj.decode("utf-8", "replace") if not isinstance(obj, string_types): obj = safe_repr(obj) is_span_description = ( len(path) == 3 and path[0] == "spans" and path[-1] == "description" ) if is_span_description: return obj return _flatten_annotated(strip_string(obj, max_length=max_value_length)) # # Start of serialize() function # disable_capture_event.set(True) try: serialized_event = _serialize_node(event, **kwargs) if meta_stack and isinstance(serialized_event, dict): serialized_event["_meta"] = meta_stack[0] return serialized_event finally: disable_capture_event.set(False)	12.97 KB	2026-01-20 12:58
📄 session.py import uuid from datetime import datetime from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.utils import format_timestamp if TYPE_CHECKING: from typing import Optional from typing import Union from typing import Any from typing import Dict from sentry_sdk._types import SessionStatus def _minute_trunc(ts): # type: (datetime) -> datetime return ts.replace(second=0, microsecond=0) def _make_uuid( val, # type: Union[str, uuid.UUID] ): # type: (...) -> uuid.UUID if isinstance(val, uuid.UUID): return val return uuid.UUID(val) class Session(object): def __init__( self, sid=None, # type: Optional[Union[str, uuid.UUID]] did=None, # type: Optional[str] timestamp=None, # type: Optional[datetime] started=None, # type: Optional[datetime] duration=None, # type: Optional[float] status=None, # type: Optional[SessionStatus] release=None, # type: Optional[str] environment=None, # type: Optional[str] user_agent=None, # type: Optional[str] ip_address=None, # type: Optional[str] errors=None, # type: Optional[int] user=None, # type: Optional[Any] session_mode="application", # type: str ): # type: (...) -> None if sid is None: sid = uuid.uuid4() if started is None: started = datetime.utcnow() if status is None: status = "ok" self.status = status self.did = None # type: Optional[str] self.started = started self.release = None # type: Optional[str] self.environment = None # type: Optional[str] self.duration = None # type: Optional[float] self.user_agent = None # type: Optional[str] self.ip_address = None # type: Optional[str] self.session_mode = session_mode # type: str self.errors = 0 self.update( sid=sid, did=did, timestamp=timestamp, duration=duration, release=release, environment=environment, user_agent=user_agent, ip_address=ip_address, errors=errors, user=user, ) @property def truncated_started(self): # type: (...) -> datetime return _minute_trunc(self.started) def update( self, sid=None, # type: Optional[Union[str, uuid.UUID]] did=None, # type: Optional[str] timestamp=None, # type: Optional[datetime] started=None, # type: Optional[datetime] duration=None, # type: Optional[float] status=None, # type: Optional[SessionStatus] release=None, # type: Optional[str] environment=None, # type: Optional[str] user_agent=None, # type: Optional[str] ip_address=None, # type: Optional[str] errors=None, # type: Optional[int] user=None, # type: Optional[Any] ): # type: (...) -> None # If a user is supplied we pull some data form it if user: if ip_address is None: ip_address = user.get("ip_address") if did is None: did = user.get("id") or user.get("email") or user.get("username") if sid is not None: self.sid = _make_uuid(sid) if did is not None: self.did = str(did) if timestamp is None: timestamp = datetime.utcnow() self.timestamp = timestamp if started is not None: self.started = started if duration is not None: self.duration = duration if release is not None: self.release = release if environment is not None: self.environment = environment if ip_address is not None: self.ip_address = ip_address if user_agent is not None: self.user_agent = user_agent if errors is not None: self.errors = errors if status is not None: self.status = status def close( self, status=None # type: Optional[SessionStatus] ): # type: (...) -> Any if status is None and self.status == "ok": status = "exited" if status is not None: self.update(status=status) def get_json_attrs( self, with_user_info=True # type: Optional[bool] ): # type: (...) -> Any attrs = {} if self.release is not None: attrs["release"] = self.release if self.environment is not None: attrs["environment"] = self.environment if with_user_info: if self.ip_address is not None: attrs["ip_address"] = self.ip_address if self.user_agent is not None: attrs["user_agent"] = self.user_agent return attrs def to_json(self): # type: (...) -> Any rv = { "sid": str(self.sid), "init": True, "started": format_timestamp(self.started), "timestamp": format_timestamp(self.timestamp), "status": self.status, } # type: Dict[str, Any] if self.errors: rv["errors"] = self.errors if self.did is not None: rv["did"] = self.did if self.duration is not None: rv["duration"] = self.duration attrs = self.get_json_attrs() if attrs: rv["attrs"] = attrs return rv	5.43 KB	2026-01-20 12:58
📄 sessions.py import os import time from threading import Thread, Lock from contextlib import contextmanager import sentry_sdk from sentry_sdk.envelope import Envelope from sentry_sdk.session import Session from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.utils import format_timestamp if TYPE_CHECKING: from typing import Any from typing import Callable from typing import Dict from typing import Generator from typing import List from typing import Optional from typing import Union def is_auto_session_tracking_enabled(hub=None): # type: (Optional[sentry_sdk.Hub]) -> Union[Any, bool, None] """Utility function to find out if session tracking is enabled.""" if hub is None: hub = sentry_sdk.Hub.current should_track = hub.scope._force_auto_session_tracking if should_track is None: client_options = hub.client.options if hub.client else {} should_track = client_options.get("auto_session_tracking", False) return should_track @contextmanager def auto_session_tracking(hub=None, session_mode="application"): # type: (Optional[sentry_sdk.Hub], str) -> Generator[None, None, None] """Starts and stops a session automatically around a block.""" if hub is None: hub = sentry_sdk.Hub.current should_track = is_auto_session_tracking_enabled(hub) if should_track: hub.start_session(session_mode=session_mode) try: yield finally: if should_track: hub.end_session() TERMINAL_SESSION_STATES = ("exited", "abnormal", "crashed") MAX_ENVELOPE_ITEMS = 100 def make_aggregate_envelope(aggregate_states, attrs): # type: (Any, Any) -> Any return {"attrs": dict(attrs), "aggregates": list(aggregate_states.values())} class SessionFlusher(object): def __init__( self, capture_func, # type: Callable[[Envelope], None] flush_interval=60, # type: int ): # type: (...) -> None self.capture_func = capture_func self.flush_interval = flush_interval self.pending_sessions = [] # type: List[Any] self.pending_aggregates = {} # type: Dict[Any, Any] self._thread = None # type: Optional[Thread] self._thread_lock = Lock() self._aggregate_lock = Lock() self._thread_for_pid = None # type: Optional[int] self._running = True def flush(self): # type: (...) -> None pending_sessions = self.pending_sessions self.pending_sessions = [] with self._aggregate_lock: pending_aggregates = self.pending_aggregates self.pending_aggregates = {} envelope = Envelope() for session in pending_sessions: if len(envelope.items) == MAX_ENVELOPE_ITEMS: self.capture_func(envelope) envelope = Envelope() envelope.add_session(session) for attrs, states in pending_aggregates.items(): if len(envelope.items) == MAX_ENVELOPE_ITEMS: self.capture_func(envelope) envelope = Envelope() envelope.add_sessions(make_aggregate_envelope(states, attrs)) if len(envelope.items) > 0: self.capture_func(envelope) def _ensure_running(self): # type: (...) -> None if self._thread_for_pid == os.getpid() and self._thread is not None: return None with self._thread_lock: if self._thread_for_pid == os.getpid() and self._thread is not None: return None def _thread(): # type: (...) -> None while self._running: time.sleep(self.flush_interval) if self._running: self.flush() thread = Thread(target=_thread) thread.daemon = True thread.start() self._thread = thread self._thread_for_pid = os.getpid() return None def add_aggregate_session( self, session # type: Session ): # type: (...) -> None # NOTE on `session.did`: # the protocol can deal with buckets that have a distinct-id, however # in practice we expect the python SDK to have an extremely high cardinality # here, effectively making aggregation useless, therefore we do not # aggregate per-did. # For this part we can get away with using the global interpreter lock with self._aggregate_lock: attrs = session.get_json_attrs(with_user_info=False) primary_key = tuple(sorted(attrs.items())) secondary_key = session.truncated_started # (, session.did) states = self.pending_aggregates.setdefault(primary_key, {}) state = states.setdefault(secondary_key, {}) if "started" not in state: state["started"] = format_timestamp(session.truncated_started) # if session.did is not None: # state["did"] = session.did if session.status == "crashed": state["crashed"] = state.get("crashed", 0) + 1 elif session.status == "abnormal": state["abnormal"] = state.get("abnormal", 0) + 1 elif session.errors > 0: state["errored"] = state.get("errored", 0) + 1 else: state["exited"] = state.get("exited", 0) + 1 def add_session( self, session # type: Session ): # type: (...) -> None if session.session_mode == "request": self.add_aggregate_session(session) else: self.pending_sessions.append(session.to_json()) self._ensure_running() def kill(self): # type: (...) -> None self._running = False def __del__(self): # type: (...) -> None self.kill()	5.76 KB	2026-01-20 12:58
📄 tracing.py import uuid import random from datetime import datetime, timedelta import sentry_sdk from sentry_sdk.consts import INSTRUMENTER from sentry_sdk.utils import is_valid_sample_rate, logger, nanosecond_time from sentry_sdk._compat import PY2 from sentry_sdk.consts import SPANDATA from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: import typing from typing import Any from typing import Dict from typing import Iterator from typing import List from typing import Optional from typing import Tuple import sentry_sdk.profiler from sentry_sdk._types import Event, MeasurementUnit, SamplingContext BAGGAGE_HEADER_NAME = "baggage" SENTRY_TRACE_HEADER_NAME = "sentry-trace" # Transaction source # see https://develop.sentry.dev/sdk/event-payloads/transaction/#transaction-annotations TRANSACTION_SOURCE_CUSTOM = "custom" TRANSACTION_SOURCE_URL = "url" TRANSACTION_SOURCE_ROUTE = "route" TRANSACTION_SOURCE_VIEW = "view" TRANSACTION_SOURCE_COMPONENT = "component" TRANSACTION_SOURCE_TASK = "task" # These are typically high cardinality and the server hates them LOW_QUALITY_TRANSACTION_SOURCES = [ TRANSACTION_SOURCE_URL, ] SOURCE_FOR_STYLE = { "endpoint": TRANSACTION_SOURCE_COMPONENT, "function_name": TRANSACTION_SOURCE_COMPONENT, "handler_name": TRANSACTION_SOURCE_COMPONENT, "method_and_path_pattern": TRANSACTION_SOURCE_ROUTE, "path": TRANSACTION_SOURCE_URL, "route_name": TRANSACTION_SOURCE_COMPONENT, "route_pattern": TRANSACTION_SOURCE_ROUTE, "uri_template": TRANSACTION_SOURCE_ROUTE, "url": TRANSACTION_SOURCE_ROUTE, } class _SpanRecorder(object): """Limits the number of spans recorded in a transaction.""" __slots__ = ("maxlen", "spans") def __init__(self, maxlen): # type: (int) -> None # FIXME: this is `maxlen - 1` only to preserve historical behavior # enforced by tests. # Either this should be changed to `maxlen` or the JS SDK implementation # should be changed to match a consistent interpretation of what maxlen # limits: either transaction+spans or only child spans. self.maxlen = maxlen - 1 self.spans = [] # type: List[Span] def add(self, span): # type: (Span) -> None if len(self.spans) > self.maxlen: span._span_recorder = None else: self.spans.append(span) class Span(object): __slots__ = ( "trace_id", "span_id", "parent_span_id", "same_process_as_parent", "sampled", "op", "description", "start_timestamp", "_start_timestamp_monotonic_ns", "status", "timestamp", "_tags", "_data", "_span_recorder", "hub", "_context_manager_state", "_containing_transaction", ) def __new__(cls, kwargs): # type: (Any) -> Any """ Backwards-compatible implementation of Span and Transaction creation. """ # TODO: consider removing this in a future release. # This is for backwards compatibility with releases before Transaction # existed, to allow for a smoother transition. if "transaction" in kwargs: return object.__new__(Transaction) return object.__new__(cls) def __init__( self, trace_id=None, # type: Optional[str] span_id=None, # type: Optional[str] parent_span_id=None, # type: Optional[str] same_process_as_parent=True, # type: bool sampled=None, # type: Optional[bool] op=None, # type: Optional[str] description=None, # type: Optional[str] hub=None, # type: Optional[sentry_sdk.Hub] status=None, # type: Optional[str] transaction=None, # type: Optional[str] # deprecated containing_transaction=None, # type: Optional[Transaction] start_timestamp=None, # type: Optional[datetime] ): # type: (...) -> None self.trace_id = trace_id or uuid.uuid4().hex self.span_id = span_id or uuid.uuid4().hex[16:] self.parent_span_id = parent_span_id self.same_process_as_parent = same_process_as_parent self.sampled = sampled self.op = op self.description = description self.status = status self.hub = hub self._tags = {} # type: Dict[str, str] self._data = {} # type: Dict[str, Any] self._containing_transaction = containing_transaction self.start_timestamp = start_timestamp or datetime.utcnow() try: # profiling depends on this value and requires that # it is measured in nanoseconds self._start_timestamp_monotonic_ns = nanosecond_time() except AttributeError: pass #: End timestamp of span self.timestamp = None # type: Optional[datetime] self._span_recorder = None # type: Optional[_SpanRecorder] # TODO this should really live on the Transaction class rather than the Span # class def init_span_recorder(self, maxlen): # type: (int) -> None if self._span_recorder is None: self._span_recorder = _SpanRecorder(maxlen) def __repr__(self): # type: () -> str return ( "<%s(op=%r, description:%r, trace_id=%r, span_id=%r, parent_span_id=%r, sampled=%r)>" % ( self.__class__.__name__, self.op, self.description, self.trace_id, self.span_id, self.parent_span_id, self.sampled, ) ) def __enter__(self): # type: () -> Span hub = self.hub or sentry_sdk.Hub.current _, scope = hub._stack[-1] old_span = scope.span scope.span = self self._context_manager_state = (hub, scope, old_span) return self def __exit__(self, ty, value, tb): # type: (Optional[Any], Optional[Any], Optional[Any]) -> None if value is not None: self.set_status("internal_error") hub, scope, old_span = self._context_manager_state del self._context_manager_state self.finish(hub) scope.span = old_span @property def containing_transaction(self): # type: () -> Optional[Transaction] # this is a getter rather than a regular attribute so that transactions # can return `self` here instead (as a way to prevent them circularly # referencing themselves) return self._containing_transaction def start_child(self, instrumenter=INSTRUMENTER.SENTRY, kwargs): # type: (str, Any) -> Span """ Start a sub-span from the current span or transaction. Takes the same arguments as the initializer of :py:class:`Span`. The trace id, sampling decision, transaction pointer, and span recorder are inherited from the current span/transaction. """ hub = self.hub or sentry_sdk.Hub.current client = hub.client configuration_instrumenter = client and client.options["instrumenter"] if instrumenter != configuration_instrumenter: return NoOpSpan() kwargs.setdefault("sampled", self.sampled) child = Span( trace_id=self.trace_id, parent_span_id=self.span_id, containing_transaction=self.containing_transaction, kwargs ) span_recorder = ( self.containing_transaction and self.containing_transaction._span_recorder ) if span_recorder: span_recorder.add(child) return child def new_span(self, kwargs): # type: (Any) -> Span """Deprecated: use :py:meth:`sentry_sdk.tracing.Span.start_child` instead.""" logger.warning("Deprecated: use Span.start_child instead of Span.new_span.") return self.start_child(kwargs) @classmethod def continue_from_environ( cls, environ, # type: typing.Mapping[str, str] kwargs # type: Any ): # type: (...) -> Transaction """ Create a Transaction with the given params, then add in data pulled from the 'sentry-trace' and 'baggage' headers from the environ (if any) before returning the Transaction. This is different from `continue_from_headers` in that it assumes header names in the form "HTTP_HEADER_NAME" - such as you would get from a wsgi environ - rather than the form "header-name". """ if cls is Span: logger.warning( "Deprecated: use Transaction.continue_from_environ " "instead of Span.continue_from_environ." ) return Transaction.continue_from_headers(EnvironHeaders(environ), kwargs) @classmethod def continue_from_headers( cls, headers, # type: typing.Mapping[str, str] kwargs # type: Any ): # type: (...) -> Transaction """ Create a transaction with the given params (including any data pulled from the 'sentry-trace' and 'baggage' headers). """ # TODO move this to the Transaction class if cls is Span: logger.warning( "Deprecated: use Transaction.continue_from_headers " "instead of Span.continue_from_headers." ) # TODO-neel move away from this kwargs stuff, it's confusing and opaque # make more explicit baggage = Baggage.from_incoming_header(headers.get(BAGGAGE_HEADER_NAME)) kwargs.update({BAGGAGE_HEADER_NAME: baggage}) sentrytrace_kwargs = extract_sentrytrace_data( headers.get(SENTRY_TRACE_HEADER_NAME) ) if sentrytrace_kwargs is not None: kwargs.update(sentrytrace_kwargs) # If there's an incoming sentry-trace but no incoming baggage header, # for instance in traces coming from older SDKs, # baggage will be empty and immutable and won't be populated as head SDK. baggage.freeze() transaction = Transaction(kwargs) transaction.same_process_as_parent = False return transaction def iter_headers(self): # type: () -> Iterator[Tuple[str, str]] """ Creates a generator which returns the span's `sentry-trace` and `baggage` headers. If the span's containing transaction doesn't yet have a `baggage` value, this will cause one to be generated and stored. """ yield SENTRY_TRACE_HEADER_NAME, self.to_traceparent() if self.containing_transaction: baggage = self.containing_transaction.get_baggage().serialize() if baggage: yield BAGGAGE_HEADER_NAME, baggage @classmethod def from_traceparent( cls, traceparent, # type: Optional[str] kwargs # type: Any ): # type: (...) -> Optional[Transaction] """ DEPRECATED: Use :py:meth:`sentry_sdk.tracing.Transaction.continue_from_headers`. Create a `Transaction` with the given params, then add in data pulled from the given 'sentry-trace' header value before returning the `Transaction`. """ logger.warning( "Deprecated: Use Transaction.continue_from_headers(headers, kwargs) " "instead of from_traceparent(traceparent, kwargs)" ) if not traceparent: return None return cls.continue_from_headers( {SENTRY_TRACE_HEADER_NAME: traceparent}, kwargs ) def to_traceparent(self): # type: () -> str if self.sampled is True: sampled = "1" elif self.sampled is False: sampled = "0" else: sampled = None traceparent = "%s-%s" % (self.trace_id, self.span_id) if sampled is not None: traceparent += "-%s" % (sampled,) return traceparent def to_baggage(self): # type: () -> Optional[Baggage] if self.containing_transaction: return self.containing_transaction.get_baggage() return None def set_tag(self, key, value): # type: (str, Any) -> None self._tags[key] = value def set_data(self, key, value): # type: (str, Any) -> None self._data[key] = value def set_status(self, value): # type: (str) -> None self.status = value def set_http_status(self, http_status): # type: (int) -> None self.set_tag( "http.status_code", str(http_status) ) # we keep this for backwards compatability self.set_data(SPANDATA.HTTP_STATUS_CODE, http_status) if http_status < 400: self.set_status("ok") elif 400 <= http_status < 500: if http_status == 403: self.set_status("permission_denied") elif http_status == 404: self.set_status("not_found") elif http_status == 429: self.set_status("resource_exhausted") elif http_status == 413: self.set_status("failed_precondition") elif http_status == 401: self.set_status("unauthenticated") elif http_status == 409: self.set_status("already_exists") else: self.set_status("invalid_argument") elif 500 <= http_status < 600: if http_status == 504: self.set_status("deadline_exceeded") elif http_status == 501: self.set_status("unimplemented") elif http_status == 503: self.set_status("unavailable") else: self.set_status("internal_error") else: self.set_status("unknown_error") def is_success(self): # type: () -> bool return self.status == "ok" def finish(self, hub=None, end_timestamp=None): # type: (Optional[sentry_sdk.Hub], Optional[datetime]) -> Optional[str] # XXX: would be type: (Optional[sentry_sdk.Hub]) -> None, but that leads # to incompatible return types for Span.finish and Transaction.finish. if self.timestamp is not None: # This span is already finished, ignore. return None hub = hub or self.hub or sentry_sdk.Hub.current try: if end_timestamp: self.timestamp = end_timestamp else: elapsed = nanosecond_time() - self._start_timestamp_monotonic_ns self.timestamp = self.start_timestamp + timedelta( microseconds=elapsed / 1000 ) except AttributeError: self.timestamp = datetime.utcnow() maybe_create_breadcrumbs_from_span(hub, self) return None def to_json(self): # type: () -> Dict[str, Any] rv = { "trace_id": self.trace_id, "span_id": self.span_id, "parent_span_id": self.parent_span_id, "same_process_as_parent": self.same_process_as_parent, "op": self.op, "description": self.description, "start_timestamp": self.start_timestamp, "timestamp": self.timestamp, } # type: Dict[str, Any] if self.status: self._tags["status"] = self.status tags = self._tags if tags: rv["tags"] = tags data = self._data if data: rv["data"] = data return rv def get_trace_context(self): # type: () -> Any rv = { "trace_id": self.trace_id, "span_id": self.span_id, "parent_span_id": self.parent_span_id, "op": self.op, "description": self.description, } # type: Dict[str, Any] if self.status: rv["status"] = self.status if self.containing_transaction: rv[ "dynamic_sampling_context" ] = self.containing_transaction.get_baggage().dynamic_sampling_context() return rv class Transaction(Span): __slots__ = ( "name", "source", "parent_sampled", # used to create baggage value for head SDKs in dynamic sampling "sample_rate", "_measurements", "_contexts", "_profile", "_baggage", ) def __init__( self, name="", # type: str parent_sampled=None, # type: Optional[bool] baggage=None, # type: Optional[Baggage] source=TRANSACTION_SOURCE_CUSTOM, # type: str kwargs # type: Any ): # type: (...) -> None # TODO: consider removing this in a future release. # This is for backwards compatibility with releases before Transaction # existed, to allow for a smoother transition. if not name and "transaction" in kwargs: logger.warning( "Deprecated: use Transaction(name=...) to create transactions " "instead of Span(transaction=...)." ) name = kwargs.pop("transaction") Span.__init__(self, kwargs) self.name = name self.source = source self.sample_rate = None # type: Optional[float] self.parent_sampled = parent_sampled self._measurements = {} # type: Dict[str, Any] self._contexts = {} # type: Dict[str, Any] self._profile = None # type: Optional[sentry_sdk.profiler.Profile] self._baggage = baggage def __repr__(self): # type: () -> str return ( "<%s(name=%r, op=%r, trace_id=%r, span_id=%r, parent_span_id=%r, sampled=%r, source=%r)>" % ( self.__class__.__name__, self.name, self.op, self.trace_id, self.span_id, self.parent_span_id, self.sampled, self.source, ) ) def __enter__(self): # type: () -> Transaction super(Transaction, self).__enter__() if self._profile is not None: self._profile.__enter__() return self def __exit__(self, ty, value, tb): # type: (Optional[Any], Optional[Any], Optional[Any]) -> None if self._profile is not None: self._profile.__exit__(ty, value, tb) super(Transaction, self).__exit__(ty, value, tb) @property def containing_transaction(self): # type: () -> Transaction # Transactions (as spans) belong to themselves (as transactions). This # is a getter rather than a regular attribute to avoid having a circular # reference. return self def finish(self, hub=None, end_timestamp=None): # type: (Optional[sentry_sdk.Hub], Optional[datetime]) -> Optional[str] if self.timestamp is not None: # This transaction is already finished, ignore. return None hub = hub or self.hub or sentry_sdk.Hub.current client = hub.client if client is None: # We have no client and therefore nowhere to send this transaction. return None # This is a de facto proxy for checking if sampled = False if self._span_recorder is None: logger.debug("Discarding transaction because sampled = False") # This is not entirely accurate because discards here are not # exclusively based on sample rate but also traces sampler, but # we handle this the same here. if client.transport and has_tracing_enabled(client.options): if client.monitor and client.monitor.downsample_factor > 1: reason = "backpressure" else: reason = "sample_rate" client.transport.record_lost_event(reason, data_category="transaction") return None if not self.name: logger.warning( "Transaction has no name, falling back to `<unlabeled transaction>`." ) self.name = "<unlabeled transaction>" Span.finish(self, hub, end_timestamp) if not self.sampled: # At this point a `sampled = None` should have already been resolved # to a concrete decision. if self.sampled is None: logger.warning("Discarding transaction without sampling decision.") return None finished_spans = [ span.to_json() for span in self._span_recorder.spans if span.timestamp is not None ] # we do this to break the circular reference of transaction -> span # recorder -> span -> containing transaction (which is where we started) # before either the spans or the transaction goes out of scope and has # to be garbage collected self._span_recorder = None contexts = {} contexts.update(self._contexts) contexts.update({"trace": self.get_trace_context()}) event = { "type": "transaction", "transaction": self.name, "transaction_info": {"source": self.source}, "contexts": contexts, "tags": self._tags, "timestamp": self.timestamp, "start_timestamp": self.start_timestamp, "spans": finished_spans, } # type: Event if self._profile is not None and self._profile.valid(): event["profile"] = self._profile self._profile = None event["measurements"] = self._measurements return hub.capture_event(event) def set_measurement(self, name, value, unit=""): # type: (str, float, MeasurementUnit) -> None self._measurements[name] = {"value": value, "unit": unit} def set_context(self, key, value): # type: (str, Any) -> None self._contexts[key] = value def to_json(self): # type: () -> Dict[str, Any] rv = super(Transaction, self).to_json() rv["name"] = self.name rv["source"] = self.source rv["sampled"] = self.sampled return rv def get_baggage(self): # type: () -> Baggage """ The first time a new baggage with sentry items is made, it will be frozen. """ if not self._baggage or self._baggage.mutable: self._baggage = Baggage.populate_from_transaction(self) return self._baggage def _set_initial_sampling_decision(self, sampling_context): # type: (SamplingContext) -> None """ Sets the transaction's sampling decision, according to the following precedence rules: 1. If a sampling decision is passed to `start_transaction` (`start_transaction(name: "my transaction", sampled: True)`), that decision will be used, regardless of anything else 2. If `traces_sampler` is defined, its decision will be used. It can choose to keep or ignore any parent sampling decision, or use the sampling context data to make its own decision or to choose a sample rate for the transaction. 3. If `traces_sampler` is not defined, but there's a parent sampling decision, the parent sampling decision will be used. 4. If `traces_sampler` is not defined and there's no parent sampling decision, `traces_sample_rate` will be used. """ hub = self.hub or sentry_sdk.Hub.current client = hub.client options = (client and client.options) or {} transaction_description = "{op}transaction <{name}>".format( op=("<" + self.op + "> " if self.op else ""), name=self.name ) # nothing to do if there's no client or if tracing is disabled if not client or not has_tracing_enabled(options): self.sampled = False return # if the user has forced a sampling decision by passing a `sampled` # value when starting the transaction, go with that if self.sampled is not None: self.sample_rate = float(self.sampled) return # we would have bailed already if neither `traces_sampler` nor # `traces_sample_rate` were defined, so one of these should work; prefer # the hook if so sample_rate = ( options["traces_sampler"](sampling_context) if callable(options.get("traces_sampler")) else ( # default inheritance behavior sampling_context["parent_sampled"] if sampling_context["parent_sampled"] is not None else options["traces_sample_rate"] ) ) # Since this is coming from the user (or from a function provided by the # user), who knows what we might get. (The only valid values are # booleans or numbers between 0 and 1.) if not is_valid_sample_rate(sample_rate, source="Tracing"): logger.warning( "[Tracing] Discarding {transaction_description} because of invalid sample rate.".format( transaction_description=transaction_description, ) ) self.sampled = False return self.sample_rate = float(sample_rate) if client.monitor: self.sample_rate /= client.monitor.downsample_factor # if the function returned 0 (or false), or if `traces_sample_rate` is # 0, it's a sign the transaction should be dropped if not self.sample_rate: logger.debug( "[Tracing] Discarding {transaction_description} because {reason}".format( transaction_description=transaction_description, reason=( "traces_sampler returned 0 or False" if callable(options.get("traces_sampler")) else "traces_sample_rate is set to 0" ), ) ) self.sampled = False return # Now we roll the dice. random.random is inclusive of 0, but not of 1, # so strict < is safe here. In case sample_rate is a boolean, cast it # to a float (True becomes 1.0 and False becomes 0.0) self.sampled = random.random() < self.sample_rate if self.sampled: logger.debug( "[Tracing] Starting {transaction_description}".format( transaction_description=transaction_description, ) ) else: logger.debug( "[Tracing] Discarding {transaction_description} because it's not included in the random sample (sampling rate = {sample_rate})".format( transaction_description=transaction_description, sample_rate=self.sample_rate, ) ) class NoOpSpan(Span): def __repr__(self): # type: () -> str return self.__class__.__name__ def start_child(self, instrumenter=INSTRUMENTER.SENTRY, kwargs): # type: (str, Any) -> NoOpSpan return NoOpSpan() def new_span(self, kwargs): # type: (Any) -> NoOpSpan return self.start_child(**kwargs) def to_traceparent(self): # type: () -> str return "" def to_baggage(self): # type: () -> Optional[Baggage] return None def iter_headers(self): # type: () -> Iterator[Tuple[str, str]] return iter(()) def set_tag(self, key, value): # type: (str, Any) -> None pass def set_data(self, key, value): # type: (str, Any) -> None pass def set_status(self, value): # type: (str) -> None pass def set_http_status(self, http_status): # type: (int) -> None pass def is_success(self): # type: () -> bool return True def to_json(self): # type: () -> Dict[str, Any] return {} def get_trace_context(self): # type: () -> Any return {} def finish(self, hub=None, end_timestamp=None): # type: (Optional[sentry_sdk.Hub], Optional[datetime]) -> Optional[str] pass def trace(func=None): # type: (Any) -> Any """ Decorator to start a child span under the existing current transaction. If there is no current transaction, then nothing will be traced. .. code-block:: :caption: Usage import sentry_sdk @sentry_sdk.trace def my_function(): ... @sentry_sdk.trace async def my_async_function(): ... """ if PY2: from sentry_sdk.tracing_utils_py2 import start_child_span_decorator else: from sentry_sdk.tracing_utils_py3 import start_child_span_decorator # This patterns allows usage of both @sentry_traced and @sentry_traced(...) # See https://stackoverflow.com/questions/52126071/decorator-with-arguments-avoid-parenthesis-when-no-arguments/52126278 if func: return start_child_span_decorator(func) else: return start_child_span_decorator # Circular imports from sentry_sdk.tracing_utils import ( Baggage, EnvironHeaders, extract_sentrytrace_data, has_tracing_enabled, maybe_create_breadcrumbs_from_span, )	29.04 KB	2026-01-20 12:58
📄 tracing_utils.py import re import contextlib import sentry_sdk from sentry_sdk.consts import OP from sentry_sdk.utils import ( capture_internal_exceptions, Dsn, match_regex_list, to_string, ) from sentry_sdk._compat import PY2, iteritems from sentry_sdk._types import TYPE_CHECKING if PY2: from collections import Mapping from urllib import quote, unquote else: from collections.abc import Mapping from urllib.parse import quote, unquote if TYPE_CHECKING: import typing from typing import Any from typing import Dict from typing import Generator from typing import Optional from typing import Union SENTRY_TRACE_REGEX = re.compile( "^[ \t]" # whitespace "([0-9a-f]{32})?" # trace_id "-?([0-9a-f]{16})?" # span_id "-?([01])?" # sampled "[ \t]$" # whitespace ) # This is a normal base64 regex, modified to reflect that fact that we strip the # trailing = or == off base64_stripped = ( # any of the characters in the base64 "alphabet", in multiples of 4 "([a-zA-Z0-9+/]{4})*" # either nothing or 2 or 3 base64-alphabet characters (see # https://en.wikipedia.org/wiki/Base64#Decoding_Base64_without_padding for # why there's never only 1 extra character) "([a-zA-Z0-9+/]{2,3})?" ) class EnvironHeaders(Mapping): # type: ignore def __init__( self, environ, # type: typing.Mapping[str, str] prefix="HTTP_", # type: str ): # type: (...) -> None self.environ = environ self.prefix = prefix def __getitem__(self, key): # type: (str) -> Optional[Any] return self.environ[self.prefix + key.replace("-", "_").upper()] def __len__(self): # type: () -> int return sum(1 for _ in iter(self)) def __iter__(self): # type: () -> Generator[str, None, None] for k in self.environ: if not isinstance(k, str): continue k = k.replace("-", "_").upper() if not k.startswith(self.prefix): continue yield k[len(self.prefix) :] def has_tracing_enabled(options): # type: (Optional[Dict[str, Any]]) -> bool """ Returns True if either traces_sample_rate or traces_sampler is defined and enable_tracing is set and not false. """ if options is None: return False return bool( options.get("enable_tracing") is not False and ( options.get("traces_sample_rate") is not None or options.get("traces_sampler") is not None ) ) @contextlib.contextmanager def record_sql_queries( hub, # type: sentry_sdk.Hub cursor, # type: Any query, # type: Any params_list, # type: Any paramstyle, # type: Optional[str] executemany, # type: bool ): # type: (...) -> Generator[sentry_sdk.tracing.Span, None, None] # TODO: Bring back capturing of params by default if hub.client and hub.client.options["_experiments"].get( "record_sql_params", False ): if not params_list or params_list == [None]: params_list = None if paramstyle == "pyformat": paramstyle = "format" else: params_list = None paramstyle = None query = _format_sql(cursor, query) data = {} if params_list is not None: data["db.params"] = params_list if paramstyle is not None: data["db.paramstyle"] = paramstyle if executemany: data["db.executemany"] = True with capture_internal_exceptions(): hub.add_breadcrumb(message=query, category="query", data=data) with hub.start_span(op=OP.DB, description=query) as span: for k, v in data.items(): span.set_data(k, v) yield span def maybe_create_breadcrumbs_from_span(hub, span): # type: (sentry_sdk.Hub, sentry_sdk.tracing.Span) -> None if span.op == OP.DB_REDIS: hub.add_breadcrumb( message=span.description, type="redis", category="redis", data=span._tags ) elif span.op == OP.HTTP_CLIENT: hub.add_breadcrumb(type="http", category="httplib", data=span._data) elif span.op == "subprocess": hub.add_breadcrumb( type="subprocess", category="subprocess", message=span.description, data=span._data, ) def extract_sentrytrace_data(header): # type: (Optional[str]) -> Optional[Dict[str, Union[str, bool, None]]] """ Given a `sentry-trace` header string, return a dictionary of data. """ if not header: return None if header.startswith("00-") and header.endswith("-00"): header = header[3:-3] match = SENTRY_TRACE_REGEX.match(header) if not match: return None trace_id, parent_span_id, sampled_str = match.groups() parent_sampled = None if trace_id: trace_id = "{:032x}".format(int(trace_id, 16)) if parent_span_id: parent_span_id = "{:016x}".format(int(parent_span_id, 16)) if sampled_str: parent_sampled = sampled_str != "0" return { "trace_id": trace_id, "parent_span_id": parent_span_id, "parent_sampled": parent_sampled, } def _format_sql(cursor, sql): # type: (Any, str) -> Optional[str] real_sql = None # If we're using psycopg2, it could be that we're # looking at a query that uses Composed objects. Use psycopg2's mogrify # function to format the query. We lose per-parameter trimming but gain # accuracy in formatting. try: if hasattr(cursor, "mogrify"): real_sql = cursor.mogrify(sql) if isinstance(real_sql, bytes): real_sql = real_sql.decode(cursor.connection.encoding) except Exception: real_sql = None return real_sql or to_string(sql) class Baggage(object): __slots__ = ("sentry_items", "third_party_items", "mutable") SENTRY_PREFIX = "sentry-" SENTRY_PREFIX_REGEX = re.compile("^sentry-") def __init__( self, sentry_items, # type: Dict[str, str] third_party_items="", # type: str mutable=True, # type: bool ): self.sentry_items = sentry_items self.third_party_items = third_party_items self.mutable = mutable @classmethod def from_incoming_header(cls, header): # type: (Optional[str]) -> Baggage """ freeze if incoming header already has sentry baggage """ sentry_items = {} third_party_items = "" mutable = True if header: for item in header.split(","): if "=" not in item: continue with capture_internal_exceptions(): item = item.strip() key, val = item.split("=") if Baggage.SENTRY_PREFIX_REGEX.match(key): baggage_key = unquote(key.split("-")[1]) sentry_items[baggage_key] = unquote(val) mutable = False else: third_party_items += ("," if third_party_items else "") + item return Baggage(sentry_items, third_party_items, mutable) @classmethod def from_options(cls, scope): # type: (sentry_sdk.scope.Scope) -> Optional[Baggage] sentry_items = {} # type: Dict[str, str] third_party_items = "" mutable = False client = sentry_sdk.Hub.current.client if client is None or scope._propagation_context is None: return Baggage(sentry_items) options = client.options propagation_context = scope._propagation_context if propagation_context is not None and "trace_id" in propagation_context: sentry_items["trace_id"] = propagation_context["trace_id"] if options.get("environment"): sentry_items["environment"] = options["environment"] if options.get("release"): sentry_items["release"] = options["release"] if options.get("dsn"): sentry_items["public_key"] = Dsn(options["dsn"]).public_key if options.get("traces_sample_rate"): sentry_items["sample_rate"] = options["traces_sample_rate"] user = (scope and scope._user) or {} if user.get("segment"): sentry_items["user_segment"] = user["segment"] return Baggage(sentry_items, third_party_items, mutable) @classmethod def populate_from_transaction(cls, transaction): # type: (sentry_sdk.tracing.Transaction) -> Baggage """ Populate fresh baggage entry with sentry_items and make it immutable if this is the head SDK which originates traces. """ hub = transaction.hub or sentry_sdk.Hub.current client = hub.client sentry_items = {} # type: Dict[str, str] if not client: return Baggage(sentry_items) options = client.options or {} user = (hub.scope and hub.scope._user) or {} sentry_items["trace_id"] = transaction.trace_id if options.get("environment"): sentry_items["environment"] = options["environment"] if options.get("release"): sentry_items["release"] = options["release"] if options.get("dsn"): sentry_items["public_key"] = Dsn(options["dsn"]).public_key if ( transaction.name and transaction.source not in LOW_QUALITY_TRANSACTION_SOURCES ): sentry_items["transaction"] = transaction.name if user.get("segment"): sentry_items["user_segment"] = user["segment"] if transaction.sample_rate is not None: sentry_items["sample_rate"] = str(transaction.sample_rate) if transaction.sampled is not None: sentry_items["sampled"] = "true" if transaction.sampled else "false" # there's an existing baggage but it was mutable, # which is why we are creating this new baggage. # However, if by chance the user put some sentry items in there, give them precedence. if transaction._baggage and transaction._baggage.sentry_items: sentry_items.update(transaction._baggage.sentry_items) return Baggage(sentry_items, mutable=False) def freeze(self): # type: () -> None self.mutable = False def dynamic_sampling_context(self): # type: () -> Dict[str, str] header = {} for key, item in iteritems(self.sentry_items): header[key] = item return header def serialize(self, include_third_party=False): # type: (bool) -> str items = [] for key, val in iteritems(self.sentry_items): with capture_internal_exceptions(): item = Baggage.SENTRY_PREFIX + quote(key) + "=" + quote(str(val)) items.append(item) if include_third_party: items.append(self.third_party_items) return ",".join(items) def should_propagate_trace(hub, url): # type: (sentry_sdk.Hub, str) -> bool """ Returns True if url matches trace_propagation_targets configured in the given hub. Otherwise, returns False. """ client = hub.client # type: Any trace_propagation_targets = client.options["trace_propagation_targets"] if client.transport and client.transport.parsed_dsn: dsn_url = client.transport.parsed_dsn.netloc else: dsn_url = None is_request_to_sentry = dsn_url and dsn_url in url if is_request_to_sentry: return False return match_regex_list(url, trace_propagation_targets, substring_matching=True) def normalize_incoming_data(incoming_data): # type: (Dict[str, Any]) -> Dict[str, Any] """ Normalizes incoming data so the keys are all lowercase with dashes instead of underscores and stripped from known prefixes. """ data = {} for key, value in incoming_data.items(): if key.startswith("HTTP_"): key = key[5:] key = key.replace("_", "-").lower() data[key] = value return data # Circular imports from sentry_sdk.tracing import LOW_QUALITY_TRANSACTION_SOURCES	12 KB	2026-01-20 12:58
📄 tracing_utils_py2.py from functools import wraps import sentry_sdk from sentry_sdk import get_current_span from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP from sentry_sdk.utils import logger, qualname_from_function if TYPE_CHECKING: from typing import Any def start_child_span_decorator(func): # type: (Any) -> Any """ Decorator to add child spans for functions. This is the Python 2 compatible version of the decorator. Duplicated code from ``sentry_sdk.tracing_utils_python3.start_child_span_decorator``. See also ``sentry_sdk.tracing.trace()``. """ @wraps(func) def func_with_tracing(args, kwargs): # type: (Any, *Any) -> Any span = get_current_span(sentry_sdk.Hub.current) if span is None: logger.warning( "Can not create a child span for %s. " "Please start a Sentry transaction before calling this function.", qualname_from_function(func), ) return func(args, *kwargs) with span.start_child( op=OP.FUNCTION, description=qualname_from_function(func), ): return func(args, **kwargs) return func_with_tracing	1.21 KB	2026-01-20 12:58
📄 tracing_utils_py3.py import inspect from functools import wraps import sentry_sdk from sentry_sdk import get_current_span from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import OP from sentry_sdk.utils import logger, qualname_from_function if TYPE_CHECKING: from typing import Any def start_child_span_decorator(func): # type: (Any) -> Any """ Decorator to add child spans for functions. This is the Python 3 compatible version of the decorator. For Python 2 there is duplicated code here: ``sentry_sdk.tracing_utils_python2.start_child_span_decorator()``. See also ``sentry_sdk.tracing.trace()``. """ # Asynchronous case if inspect.iscoroutinefunction(func): @wraps(func) async def func_with_tracing(args, kwargs): # type: (Any, *Any) -> Any span = get_current_span(sentry_sdk.Hub.current) if span is None: logger.warning( "Can not create a child span for %s. " "Please start a Sentry transaction before calling this function.", qualname_from_function(func), ) return await func(args, *kwargs) with span.start_child( op=OP.FUNCTION, description=qualname_from_function(func), ): return await func(args, *kwargs) # Synchronous case else: @wraps(func) def func_with_tracing(args, *kwargs): # type: (Any, *Any) -> Any span = get_current_span(sentry_sdk.Hub.current) if span is None: logger.warning( "Can not create a child span for %s. " "Please start a Sentry transaction before calling this function.", qualname_from_function(func), ) return func(args, *kwargs) with span.start_child( op=OP.FUNCTION, description=qualname_from_function(func), ): return func(args, **kwargs) return func_with_tracing	2.1 KB	2026-01-20 12:58
📄 transport.py from __future__ import print_function import io import urllib3 import certifi import gzip import time from datetime import datetime, timedelta from collections import defaultdict from sentry_sdk.utils import Dsn, logger, capture_internal_exceptions, json_dumps from sentry_sdk.worker import BackgroundWorker from sentry_sdk.envelope import Envelope, Item, PayloadRef from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Any from typing import Callable from typing import Dict from typing import Iterable from typing import Optional from typing import Tuple from typing import Type from typing import Union from typing import DefaultDict from urllib3.poolmanager import PoolManager from urllib3.poolmanager import ProxyManager from sentry_sdk._types import Event, EndpointType DataCategory = Optional[str] try: from urllib.request import getproxies except ImportError: from urllib import getproxies # type: ignore class Transport(object): """Baseclass for all transports. A transport is used to send an event to sentry. """ parsed_dsn = None # type: Optional[Dsn] def __init__( self, options=None # type: Optional[Dict[str, Any]] ): # type: (...) -> None self.options = options if options and options["dsn"] is not None and options["dsn"]: self.parsed_dsn = Dsn(options["dsn"]) else: self.parsed_dsn = None def capture_event( self, event # type: Event ): # type: (...) -> None """ This gets invoked with the event dictionary when an event should be sent to sentry. """ raise NotImplementedError() def capture_envelope( self, envelope # type: Envelope ): # type: (...) -> None """ Send an envelope to Sentry. Envelopes are a data container format that can hold any type of data submitted to Sentry. We use it for transactions and sessions, but regular "error" events should go through `capture_event` for backwards compat. """ raise NotImplementedError() def flush( self, timeout, # type: float callback=None, # type: Optional[Any] ): # type: (...) -> None """Wait `timeout` seconds for the current events to be sent out.""" pass def kill(self): # type: () -> None """Forcefully kills the transport.""" pass def record_lost_event( self, reason, # type: str data_category=None, # type: Optional[str] item=None, # type: Optional[Item] ): # type: (...) -> None """This increments a counter for event loss by reason and data category. """ return None def is_healthy(self): # type: () -> bool return True def __del__(self): # type: () -> None try: self.kill() except Exception: pass def _parse_rate_limits(header, now=None): # type: (Any, Optional[datetime]) -> Iterable[Tuple[DataCategory, datetime]] if now is None: now = datetime.utcnow() for limit in header.split(","): try: retry_after, categories, _ = limit.strip().split(":", 2) retry_after = now + timedelta(seconds=int(retry_after)) for category in categories and categories.split(";") or (None,): yield category, retry_after except (LookupError, ValueError): continue class HttpTransport(Transport): """The default HTTP transport.""" def __init__( self, options # type: Dict[str, Any] ): # type: (...) -> None from sentry_sdk.consts import VERSION Transport.__init__(self, options) assert self.parsed_dsn is not None self.options = options # type: Dict[str, Any] self._worker = BackgroundWorker(queue_size=options["transport_queue_size"]) self._auth = self.parsed_dsn.to_auth("sentry.python/%s" % VERSION) self._disabled_until = {} # type: Dict[DataCategory, datetime] self._retry = urllib3.util.Retry() self._discarded_events = defaultdict( int ) # type: DefaultDict[Tuple[str, str], int] self._last_client_report_sent = time.time() self._pool = self._make_pool( self.parsed_dsn, http_proxy=options["http_proxy"], https_proxy=options["https_proxy"], ca_certs=options["ca_certs"], proxy_headers=options["proxy_headers"], ) from sentry_sdk import Hub self.hub_cls = Hub def record_lost_event( self, reason, # type: str data_category=None, # type: Optional[str] item=None, # type: Optional[Item] ): # type: (...) -> None if not self.options["send_client_reports"]: return quantity = 1 if item is not None: data_category = item.data_category if data_category == "attachment": # quantity of 0 is actually 1 as we do not want to count # empty attachments as actually empty. quantity = len(item.get_bytes()) or 1 elif data_category is None: raise TypeError("data category not provided") self._discarded_events[data_category, reason] += quantity def _update_rate_limits(self, response): # type: (urllib3.BaseHTTPResponse) -> None # new sentries with more rate limit insights. We honor this header # no matter of the status code to update our internal rate limits. header = response.headers.get("x-sentry-rate-limits") if header: logger.warning("Rate-limited via x-sentry-rate-limits") self._disabled_until.update(_parse_rate_limits(header)) # old sentries only communicate global rate limit hits via the # retry-after header on 429. This header can also be emitted on new # sentries if a proxy in front wants to globally slow things down. elif response.status == 429: logger.warning("Rate-limited via 429") self._disabled_until[None] = datetime.utcnow() + timedelta( seconds=self._retry.get_retry_after(response) or 60 ) def _send_request( self, body, # type: bytes headers, # type: Dict[str, str] endpoint_type="store", # type: EndpointType envelope=None, # type: Optional[Envelope] ): # type: (...) -> None def record_loss(reason): # type: (str) -> None if envelope is None: self.record_lost_event(reason, data_category="error") else: for item in envelope.items: self.record_lost_event(reason, item=item) headers.update( { "User-Agent": str(self._auth.client), "X-Sentry-Auth": str(self._auth.to_header()), } ) try: response = self._pool.request( "POST", str(self._auth.get_api_url(endpoint_type)), body=body, headers=headers, ) except Exception: self.on_dropped_event("network") record_loss("network_error") raise try: self._update_rate_limits(response) if response.status == 429: # if we hit a 429. Something was rate limited but we already # acted on this in `self._update_rate_limits`. Note that we # do not want to record event loss here as we will have recorded # an outcome in relay already. self.on_dropped_event("status_429") pass elif response.status >= 300 or response.status < 200: logger.error( "Unexpected status code: %s (body: %s)", response.status, response.data, ) self.on_dropped_event("status_{}".format(response.status)) record_loss("network_error") finally: response.close() def on_dropped_event(self, reason): # type: (str) -> None return None def _fetch_pending_client_report(self, force=False, interval=60): # type: (bool, int) -> Optional[Item] if not self.options["send_client_reports"]: return None if not (force or self._last_client_report_sent < time.time() - interval): return None discarded_events = self._discarded_events self._discarded_events = defaultdict(int) self._last_client_report_sent = time.time() if not discarded_events: return None return Item( PayloadRef( json={ "timestamp": time.time(), "discarded_events": [ {"reason": reason, "category": category, "quantity": quantity} for ( (category, reason), quantity, ) in discarded_events.items() ], } ), type="client_report", ) def _flush_client_reports(self, force=False): # type: (bool) -> None client_report = self._fetch_pending_client_report(force=force, interval=60) if client_report is not None: self.capture_envelope(Envelope(items=[client_report])) def _check_disabled(self, category): # type: (str) -> bool def _disabled(bucket): # type: (Any) -> bool ts = self._disabled_until.get(bucket) return ts is not None and ts > datetime.utcnow() return _disabled(category) or _disabled(None) def _is_rate_limited(self): # type: () -> bool return any(ts > datetime.utcnow() for ts in self._disabled_until.values()) def _is_worker_full(self): # type: () -> bool return self._worker.full() def is_healthy(self): # type: () -> bool return not (self._is_worker_full() or self._is_rate_limited()) def _send_event( self, event # type: Event ): # type: (...) -> None if self._check_disabled("error"): self.on_dropped_event("self_rate_limits") self.record_lost_event("ratelimit_backoff", data_category="error") return None body = io.BytesIO() with gzip.GzipFile(fileobj=body, mode="w") as f: f.write(json_dumps(event)) assert self.parsed_dsn is not None logger.debug( "Sending event, type:%s level:%s event_id:%s project:%s host:%s" % ( event.get("type") or "null", event.get("level") or "null", event.get("event_id") or "null", self.parsed_dsn.project_id, self.parsed_dsn.host, ) ) self._send_request( body.getvalue(), headers={"Content-Type": "application/json", "Content-Encoding": "gzip"}, ) return None def _send_envelope( self, envelope # type: Envelope ): # type: (...) -> None # remove all items from the envelope which are over quota new_items = [] for item in envelope.items: if self._check_disabled(item.data_category): if item.data_category in ("transaction", "error", "default"): self.on_dropped_event("self_rate_limits") self.record_lost_event("ratelimit_backoff", item=item) else: new_items.append(item) # Since we're modifying the envelope here make a copy so that others # that hold references do not see their envelope modified. envelope = Envelope(headers=envelope.headers, items=new_items) if not envelope.items: return None # since we're already in the business of sending out an envelope here # check if we have one pending for the stats session envelopes so we # can attach it to this enveloped scheduled for sending. This will # currently typically attach the client report to the most recent # session update. client_report_item = self._fetch_pending_client_report(interval=30) if client_report_item is not None: envelope.items.append(client_report_item) body = io.BytesIO() with gzip.GzipFile(fileobj=body, mode="w") as f: envelope.serialize_into(f) assert self.parsed_dsn is not None logger.debug( "Sending envelope [%s] project:%s host:%s", envelope.description, self.parsed_dsn.project_id, self.parsed_dsn.host, ) self._send_request( body.getvalue(), headers={ "Content-Type": "application/x-sentry-envelope", "Content-Encoding": "gzip", }, endpoint_type="envelope", envelope=envelope, ) return None def _get_pool_options(self, ca_certs): # type: (Optional[Any]) -> Dict[str, Any] return { "num_pools": 2, "cert_reqs": "CERT_REQUIRED", "ca_certs": ca_certs or certifi.where(), } def _in_no_proxy(self, parsed_dsn): # type: (Dsn) -> bool no_proxy = getproxies().get("no") if not no_proxy: return False for host in no_proxy.split(","): host = host.strip() if parsed_dsn.host.endswith(host) or parsed_dsn.netloc.endswith(host): return True return False def _make_pool( self, parsed_dsn, # type: Dsn http_proxy, # type: Optional[str] https_proxy, # type: Optional[str] ca_certs, # type: Optional[Any] proxy_headers, # type: Optional[Dict[str, str]] ): # type: (...) -> Union[PoolManager, ProxyManager] proxy = None no_proxy = self._in_no_proxy(parsed_dsn) # try HTTPS first if parsed_dsn.scheme == "https" and (https_proxy != ""): proxy = https_proxy or (not no_proxy and getproxies().get("https")) # maybe fallback to HTTP proxy if not proxy and (http_proxy != ""): proxy = http_proxy or (not no_proxy and getproxies().get("http")) opts = self._get_pool_options(ca_certs) if proxy: if proxy_headers: opts["proxy_headers"] = proxy_headers if proxy.startswith("socks"): use_socks_proxy = True try: # Check if PySocks depencency is available from urllib3.contrib.socks import SOCKSProxyManager except ImportError: use_socks_proxy = False logger.warning( "You have configured a SOCKS proxy (%s) but support for SOCKS proxies is not installed. Disabling proxy support. Please add `PySocks` (or `urllib3` with the `[socks]` extra) to your dependencies.", proxy, ) if use_socks_proxy: return SOCKSProxyManager(proxy, opts) else: return urllib3.PoolManager(opts) else: return urllib3.ProxyManager(proxy, opts) else: return urllib3.PoolManager(opts) def capture_event( self, event # type: Event ): # type: (...) -> None hub = self.hub_cls.current def send_event_wrapper(): # type: () -> None with hub: with capture_internal_exceptions(): self._send_event(event) self._flush_client_reports() if not self._worker.submit(send_event_wrapper): self.on_dropped_event("full_queue") self.record_lost_event("queue_overflow", data_category="error") def capture_envelope( self, envelope # type: Envelope ): # type: (...) -> None hub = self.hub_cls.current def send_envelope_wrapper(): # type: () -> None with hub: with capture_internal_exceptions(): self._send_envelope(envelope) self._flush_client_reports() if not self._worker.submit(send_envelope_wrapper): self.on_dropped_event("full_queue") for item in envelope.items: self.record_lost_event("queue_overflow", item=item) def flush( self, timeout, # type: float callback=None, # type: Optional[Any] ): # type: (...) -> None logger.debug("Flushing HTTP transport") if timeout > 0: self._worker.submit(lambda: self._flush_client_reports(force=True)) self._worker.flush(timeout, callback) def kill(self): # type: () -> None logger.debug("Killing HTTP transport") self._worker.kill() class _FunctionTransport(Transport): def __init__( self, func # type: Callable[[Event], None] ): # type: (...) -> None Transport.__init__(self) self._func = func def capture_event( self, event # type: Event ): # type: (...) -> None self._func(event) return None def make_transport(options): # type: (Dict[str, Any]) -> Optional[Transport] ref_transport = options["transport"] # If no transport is given, we use the http transport class if ref_transport is None: transport_cls = HttpTransport # type: Type[Transport] elif isinstance(ref_transport, Transport): return ref_transport elif isinstance(ref_transport, type) and issubclass(ref_transport, Transport): transport_cls = ref_transport elif callable(ref_transport): return _FunctionTransport(ref_transport) # type: ignore # if a transport class is given only instantiate it if the dsn is not # empty or None if options["dsn"]: return transport_cls(options) return None	18.13 KB	2026-01-20 12:58
📄 utils.py import base64 import json import linecache import logging import math import os import re import subprocess import sys import threading import time from collections import namedtuple from copy import copy from decimal import Decimal from numbers import Real try: # Python 3 from urllib.parse import parse_qs from urllib.parse import unquote from urllib.parse import urlencode from urllib.parse import urlsplit from urllib.parse import urlunsplit except ImportError: # Python 2 from cgi import parse_qs # type: ignore from urllib import unquote # type: ignore from urllib import urlencode # type: ignore from urlparse import urlsplit # type: ignore from urlparse import urlunsplit # type: ignore try: # Python 3.11 from builtins import BaseExceptionGroup except ImportError: # Python 3.10 and below BaseExceptionGroup = None # type: ignore from datetime import datetime from functools import partial try: from functools import partialmethod _PARTIALMETHOD_AVAILABLE = True except ImportError: _PARTIALMETHOD_AVAILABLE = False import sentry_sdk from sentry_sdk._compat import PY2, PY33, PY37, implements_str, text_type, urlparse from sentry_sdk._types import TYPE_CHECKING from sentry_sdk.consts import DEFAULT_MAX_VALUE_LENGTH if TYPE_CHECKING: from types import FrameType, TracebackType from typing import ( Any, Callable, ContextManager, Dict, Iterator, List, Optional, Set, Tuple, Type, Union, ) from sentry_sdk._types import EndpointType, ExcInfo epoch = datetime(1970, 1, 1) # The logger is created here but initialized in the debug support module logger = logging.getLogger("sentry_sdk.errors") BASE64_ALPHABET = re.compile(r"^[a-zA-Z0-9/+=]$") SENSITIVE_DATA_SUBSTITUTE = "[Filtered]" def json_dumps(data): # type: (Any) -> bytes """Serialize data into a compact JSON representation encoded as UTF-8.""" return json.dumps(data, allow_nan=False, separators=(",", ":")).encode("utf-8") def _get_debug_hub(): # type: () -> Optional[sentry_sdk.Hub] # This function is replaced by debug.py pass def get_default_release(): # type: () -> Optional[str] """Try to guess a default release.""" release = os.environ.get("SENTRY_RELEASE") if release: return release with open(os.path.devnull, "w+") as null: try: release = ( subprocess.Popen( ["git", "rev-parse", "HEAD"], stdout=subprocess.PIPE, stderr=null, stdin=null, ) .communicate()[0] .strip() .decode("utf-8") ) except (OSError, IOError): pass if release: return release for var in ( "HEROKU_SLUG_COMMIT", "SOURCE_VERSION", "CODEBUILD_RESOLVED_SOURCE_VERSION", "CIRCLE_SHA1", "GAE_DEPLOYMENT_ID", ): release = os.environ.get(var) if release: return release return None def get_sdk_name(installed_integrations): # type: (List[str]) -> str """Return the SDK name including the name of the used web framework.""" # Note: I can not use for example sentry_sdk.integrations.django.DjangoIntegration.identifier # here because if django is not installed the integration is not accessible. framework_integrations = [ "django", "flask", "fastapi", "bottle", "falcon", "quart", "sanic", "starlette", "chalice", "serverless", "pyramid", "tornado", "aiohttp", "aws_lambda", "gcp", "beam", "asgi", "wsgi", ] for integration in framework_integrations: if integration in installed_integrations: return "sentry.python.{}".format(integration) return "sentry.python" class CaptureInternalException(object): __slots__ = () def __enter__(self): # type: () -> ContextManager[Any] return self def __exit__(self, ty, value, tb): # type: (Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]) -> bool if ty is not None and value is not None: capture_internal_exception((ty, value, tb)) return True _CAPTURE_INTERNAL_EXCEPTION = CaptureInternalException() def capture_internal_exceptions(): # type: () -> ContextManager[Any] return _CAPTURE_INTERNAL_EXCEPTION def capture_internal_exception(exc_info): # type: (ExcInfo) -> None hub = _get_debug_hub() if hub is not None: hub._capture_internal_exception(exc_info) def to_timestamp(value): # type: (datetime) -> float return (value - epoch).total_seconds() def format_timestamp(value): # type: (datetime) -> str return value.strftime("%Y-%m-%dT%H:%M:%S.%fZ") def event_hint_with_exc_info(exc_info=None): # type: (Optional[ExcInfo]) -> Dict[str, Optional[ExcInfo]] """Creates a hint with the exc info filled in.""" if exc_info is None: exc_info = sys.exc_info() else: exc_info = exc_info_from_error(exc_info) if exc_info[0] is None: exc_info = None return {"exc_info": exc_info} class BadDsn(ValueError): """Raised on invalid DSNs.""" @implements_str class Dsn(object): """Represents a DSN.""" def __init__(self, value): # type: (Union[Dsn, str]) -> None if isinstance(value, Dsn): self.__dict__ = dict(value.__dict__) return parts = urlparse.urlsplit(text_type(value)) if parts.scheme not in ("http", "https"): raise BadDsn("Unsupported scheme %r" % parts.scheme) self.scheme = parts.scheme if parts.hostname is None: raise BadDsn("Missing hostname") self.host = parts.hostname if parts.port is None: self.port = self.scheme == "https" and 443 or 80 # type: int else: self.port = parts.port if not parts.username: raise BadDsn("Missing public key") self.public_key = parts.username self.secret_key = parts.password path = parts.path.rsplit("/", 1) try: self.project_id = text_type(int(path.pop())) except (ValueError, TypeError): raise BadDsn("Invalid project in DSN (%r)" % (parts.path or "")[1:]) self.path = "/".join(path) + "/" @property def netloc(self): # type: () -> str """The netloc part of a DSN.""" rv = self.host if (self.scheme, self.port) not in (("http", 80), ("https", 443)): rv = "%s:%s" % (rv, self.port) return rv def to_auth(self, client=None): # type: (Optional[Any]) -> Auth """Returns the auth info object for this dsn.""" return Auth( scheme=self.scheme, host=self.netloc, path=self.path, project_id=self.project_id, public_key=self.public_key, secret_key=self.secret_key, client=client, ) def __str__(self): # type: () -> str return "%s://%s%s@%s%s%s" % ( self.scheme, self.public_key, self.secret_key and "@" + self.secret_key or "", self.netloc, self.path, self.project_id, ) class Auth(object): """Helper object that represents the auth info.""" def __init__( self, scheme, host, project_id, public_key, secret_key=None, version=7, client=None, path="/", ): # type: (str, str, str, str, Optional[str], int, Optional[Any], str) -> None self.scheme = scheme self.host = host self.path = path self.project_id = project_id self.public_key = public_key self.secret_key = secret_key self.version = version self.client = client @property def store_api_url(self): # type: () -> str """Returns the API url for storing events. Deprecated: use get_api_url instead. """ return self.get_api_url(type="store") def get_api_url( self, type="store" # type: EndpointType ): # type: (...) -> str """Returns the API url for storing events.""" return "%s://%s%sapi/%s/%s/" % ( self.scheme, self.host, self.path, self.project_id, type, ) def to_header(self): # type: () -> str """Returns the auth header a string.""" rv = [("sentry_key", self.public_key), ("sentry_version", self.version)] if self.client is not None: rv.append(("sentry_client", self.client)) if self.secret_key is not None: rv.append(("sentry_secret", self.secret_key)) return "Sentry " + ", ".join("%s=%s" % (key, value) for key, value in rv) class AnnotatedValue(object): """ Meta information for a data field in the event payload. This is to tell Relay that we have tampered with the fields value. See: https://github.com/getsentry/relay/blob/be12cd49a0f06ea932ed9b9f93a655de5d6ad6d1/relay-general/src/types/meta.rs#L407-L423 """ __slots__ = ("value", "metadata") def __init__(self, value, metadata): # type: (Optional[Any], Dict[str, Any]) -> None self.value = value self.metadata = metadata @classmethod def removed_because_raw_data(cls): # type: () -> AnnotatedValue """The value was removed because it could not be parsed. This is done for request body values that are not json nor a form.""" return AnnotatedValue( value="", metadata={ "rem": [ # Remark [ "!raw", # Unparsable raw data "x", # The fields original value was removed ] ] }, ) @classmethod def removed_because_over_size_limit(cls): # type: () -> AnnotatedValue """The actual value was removed because the size of the field exceeded the configured maximum size (specified with the max_request_body_size sdk option)""" return AnnotatedValue( value="", metadata={ "rem": [ # Remark [ "!config", # Because of configured maximum size "x", # The fields original value was removed ] ] }, ) @classmethod def substituted_because_contains_sensitive_data(cls): # type: () -> AnnotatedValue """The actual value was removed because it contained sensitive information.""" return AnnotatedValue( value=SENSITIVE_DATA_SUBSTITUTE, metadata={ "rem": [ # Remark [ "!config", # Because of SDK configuration (in this case the config is the hard coded removal of certain django cookies) "s", # The fields original value was substituted ] ] }, ) if TYPE_CHECKING: from typing import TypeVar T = TypeVar("T") Annotated = Union[AnnotatedValue, T] def get_type_name(cls): # type: (Optional[type]) -> Optional[str] return getattr(cls, "__qualname__", None) or getattr(cls, "__name__", None) def get_type_module(cls): # type: (Optional[type]) -> Optional[str] mod = getattr(cls, "__module__", None) if mod not in (None, "builtins", "__builtins__"): return mod return None def should_hide_frame(frame): # type: (FrameType) -> bool try: mod = frame.f_globals["__name__"] if mod.startswith("sentry_sdk."): return True except (AttributeError, KeyError): pass for flag_name in "__traceback_hide__", "__tracebackhide__": try: if frame.f_locals[flag_name]: return True except Exception: pass return False def iter_stacks(tb): # type: (Optional[TracebackType]) -> Iterator[TracebackType] tb_ = tb # type: Optional[TracebackType] while tb_ is not None: if not should_hide_frame(tb_.tb_frame): yield tb_ tb_ = tb_.tb_next def get_lines_from_file( filename, # type: str lineno, # type: int max_length=None, # type: Optional[int] loader=None, # type: Optional[Any] module=None, # type: Optional[str] ): # type: (...) -> Tuple[List[Annotated[str]], Optional[Annotated[str]], List[Annotated[str]]] context_lines = 5 source = None if loader is not None and hasattr(loader, "get_source"): try: source_str = loader.get_source(module) # type: Optional[str] except (ImportError, IOError): source_str = None if source_str is not None: source = source_str.splitlines() if source is None: try: source = linecache.getlines(filename) except (OSError, IOError): return [], None, [] if not source: return [], None, [] lower_bound = max(0, lineno - context_lines) upper_bound = min(lineno + 1 + context_lines, len(source)) try: pre_context = [ strip_string(line.strip("\r\n"), max_length=max_length) for line in source[lower_bound:lineno] ] context_line = strip_string(source[lineno].strip("\r\n"), max_length=max_length) post_context = [ strip_string(line.strip("\r\n"), max_length=max_length) for line in source[(lineno + 1) : upper_bound] ] return pre_context, context_line, post_context except IndexError: # the file may have changed since it was loaded into memory return [], None, [] def get_source_context( frame, # type: FrameType tb_lineno, # type: int max_value_length=None, # type: Optional[int] ): # type: (...) -> Tuple[List[Annotated[str]], Optional[Annotated[str]], List[Annotated[str]]] try: abs_path = frame.f_code.co_filename # type: Optional[str] except Exception: abs_path = None try: module = frame.f_globals["__name__"] except Exception: return [], None, [] try: loader = frame.f_globals["__loader__"] except Exception: loader = None lineno = tb_lineno - 1 if lineno is not None and abs_path: return get_lines_from_file( abs_path, lineno, max_value_length, loader=loader, module=module ) return [], None, [] def safe_str(value): # type: (Any) -> str try: return text_type(value) except Exception: return safe_repr(value) if PY2: def safe_repr(value): # type: (Any) -> str try: rv = repr(value).decode("utf-8", "replace") # At this point `rv` contains a bunch of literal escape codes, like # this (exaggerated example): # # u"\\x2f" # # But we want to show this string as: # # u"/" try: # unicode-escape does this job, but can only decode latin1. So we # attempt to encode in latin1. return rv.encode("latin1").decode("unicode-escape") except Exception: # Since usually strings aren't latin1 this can break. In those # cases we just give up. return rv except Exception: # If e.g. the call to `repr` already fails return "<broken repr>" else: def safe_repr(value): # type: (Any) -> str try: return repr(value) except Exception: return "<broken repr>" def filename_for_module(module, abs_path): # type: (Optional[str], Optional[str]) -> Optional[str] if not abs_path or not module: return abs_path try: if abs_path.endswith(".pyc"): abs_path = abs_path[:-1] base_module = module.split(".", 1)[0] if base_module == module: return os.path.basename(abs_path) base_module_path = sys.modules[base_module].__file__ if not base_module_path: return abs_path return abs_path.split(base_module_path.rsplit(os.sep, 2)[0], 1)[-1].lstrip( os.sep ) except Exception: return abs_path def serialize_frame( frame, tb_lineno=None, include_local_variables=True, include_source_context=True, max_value_length=None, ): # type: (FrameType, Optional[int], bool, bool, Optional[int]) -> Dict[str, Any] f_code = getattr(frame, "f_code", None) if not f_code: abs_path = None function = None else: abs_path = frame.f_code.co_filename function = frame.f_code.co_name try: module = frame.f_globals["__name__"] except Exception: module = None if tb_lineno is None: tb_lineno = frame.f_lineno rv = { "filename": filename_for_module(module, abs_path) or None, "abs_path": os.path.abspath(abs_path) if abs_path else None, "function": function or "<unknown>", "module": module, "lineno": tb_lineno, } # type: Dict[str, Any] if include_source_context: rv["pre_context"], rv["context_line"], rv["post_context"] = get_source_context( frame, tb_lineno, max_value_length ) if include_local_variables: rv["vars"] = copy(frame.f_locals) return rv def current_stacktrace( include_local_variables=True, # type: bool include_source_context=True, # type: bool max_value_length=None, # type: Optional[int] ): # type: (...) -> Dict[str, Any] __tracebackhide__ = True frames = [] f = sys._getframe() # type: Optional[FrameType] while f is not None: if not should_hide_frame(f): frames.append( serialize_frame( f, include_local_variables=include_local_variables, include_source_context=include_source_context, max_value_length=max_value_length, ) ) f = f.f_back frames.reverse() return {"frames": frames} def get_errno(exc_value): # type: (BaseException) -> Optional[Any] return getattr(exc_value, "errno", None) def single_exception_from_error_tuple( exc_type, # type: Optional[type] exc_value, # type: Optional[BaseException] tb, # type: Optional[TracebackType] client_options=None, # type: Optional[Dict[str, Any]] mechanism=None, # type: Optional[Dict[str, Any]] exception_id=None, # type: Optional[int] parent_id=None, # type: Optional[int] source=None, # type: Optional[str] ): # type: (...) -> Dict[str, Any] """ Creates a dict that goes into the events `exception.values` list and is ingestible by Sentry. See the Exception Interface documentation for more details: https://develop.sentry.dev/sdk/event-payloads/exception/ """ exception_value = {} # type: Dict[str, Any] exception_value["mechanism"] = ( mechanism.copy() if mechanism else {"type": "generic", "handled": True} ) if exception_id is not None: exception_value["mechanism"]["exception_id"] = exception_id if exc_value is not None: errno = get_errno(exc_value) else: errno = None if errno is not None: exception_value["mechanism"].setdefault("meta", {}).setdefault( "errno", {} ).setdefault("number", errno) if source is not None: exception_value["mechanism"]["source"] = source is_root_exception = exception_id == 0 if not is_root_exception and parent_id is not None: exception_value["mechanism"]["parent_id"] = parent_id exception_value["mechanism"]["type"] = "chained" if is_root_exception and "type" not in exception_value["mechanism"]: exception_value["mechanism"]["type"] = "generic" is_exception_group = BaseExceptionGroup is not None and isinstance( exc_value, BaseExceptionGroup ) if is_exception_group: exception_value["mechanism"]["is_exception_group"] = True exception_value["module"] = get_type_module(exc_type) exception_value["type"] = get_type_name(exc_type) exception_value["value"] = getattr(exc_value, "message", safe_str(exc_value)) if client_options is None: include_local_variables = True include_source_context = True max_value_length = DEFAULT_MAX_VALUE_LENGTH # fallback else: include_local_variables = client_options["include_local_variables"] include_source_context = client_options["include_source_context"] max_value_length = client_options["max_value_length"] frames = [ serialize_frame( tb.tb_frame, tb_lineno=tb.tb_lineno, include_local_variables=include_local_variables, include_source_context=include_source_context, max_value_length=max_value_length, ) for tb in iter_stacks(tb) ] if frames: exception_value["stacktrace"] = {"frames": frames} return exception_value HAS_CHAINED_EXCEPTIONS = hasattr(Exception, "__suppress_context__") if HAS_CHAINED_EXCEPTIONS: def walk_exception_chain(exc_info): # type: (ExcInfo) -> Iterator[ExcInfo] exc_type, exc_value, tb = exc_info seen_exceptions = [] seen_exception_ids = set() # type: Set[int] while ( exc_type is not None and exc_value is not None and id(exc_value) not in seen_exception_ids ): yield exc_type, exc_value, tb # Avoid hashing random types we don't know anything # about. Use the list to keep a ref so that the `id` is # not used for another object. seen_exceptions.append(exc_value) seen_exception_ids.add(id(exc_value)) if exc_value.__suppress_context__: cause = exc_value.__cause__ else: cause = exc_value.__context__ if cause is None: break exc_type = type(cause) exc_value = cause tb = getattr(cause, "__traceback__", None) else: def walk_exception_chain(exc_info): # type: (ExcInfo) -> Iterator[ExcInfo] yield exc_info def exceptions_from_error( exc_type, # type: Optional[type] exc_value, # type: Optional[BaseException] tb, # type: Optional[TracebackType] client_options=None, # type: Optional[Dict[str, Any]] mechanism=None, # type: Optional[Dict[str, Any]] exception_id=0, # type: int parent_id=0, # type: int source=None, # type: Optional[str] ): # type: (...) -> Tuple[int, List[Dict[str, Any]]] """ Creates the list of exceptions. This can include chained exceptions and exceptions from an ExceptionGroup. See the Exception Interface documentation for more details: https://develop.sentry.dev/sdk/event-payloads/exception/ """ parent = single_exception_from_error_tuple( exc_type=exc_type, exc_value=exc_value, tb=tb, client_options=client_options, mechanism=mechanism, exception_id=exception_id, parent_id=parent_id, source=source, ) exceptions = [parent] parent_id = exception_id exception_id += 1 should_supress_context = hasattr(exc_value, "__suppress_context__") and exc_value.__suppress_context__ # type: ignore if should_supress_context: # Add direct cause. # The field `__cause__` is set when raised with the exception (using the `from` keyword). exception_has_cause = ( exc_value and hasattr(exc_value, "__cause__") and exc_value.__cause__ is not None ) if exception_has_cause: cause = exc_value.__cause__ # type: ignore (exception_id, child_exceptions) = exceptions_from_error( exc_type=type(cause), exc_value=cause, tb=getattr(cause, "__traceback__", None), client_options=client_options, mechanism=mechanism, exception_id=exception_id, source="__cause__", ) exceptions.extend(child_exceptions) else: # Add indirect cause. # The field `__context__` is assigned if another exception occurs while handling the exception. exception_has_content = ( exc_value and hasattr(exc_value, "__context__") and exc_value.__context__ is not None ) if exception_has_content: context = exc_value.__context__ # type: ignore (exception_id, child_exceptions) = exceptions_from_error( exc_type=type(context), exc_value=context, tb=getattr(context, "__traceback__", None), client_options=client_options, mechanism=mechanism, exception_id=exception_id, source="__context__", ) exceptions.extend(child_exceptions) # Add exceptions from an ExceptionGroup. is_exception_group = exc_value and hasattr(exc_value, "exceptions") if is_exception_group: for idx, e in enumerate(exc_value.exceptions): # type: ignore (exception_id, child_exceptions) = exceptions_from_error( exc_type=type(e), exc_value=e, tb=getattr(e, "__traceback__", None), client_options=client_options, mechanism=mechanism, exception_id=exception_id, parent_id=parent_id, source="exceptions[%s]" % idx, ) exceptions.extend(child_exceptions) return (exception_id, exceptions) def exceptions_from_error_tuple( exc_info, # type: ExcInfo client_options=None, # type: Optional[Dict[str, Any]] mechanism=None, # type: Optional[Dict[str, Any]] ): # type: (...) -> List[Dict[str, Any]] exc_type, exc_value, tb = exc_info is_exception_group = BaseExceptionGroup is not None and isinstance( exc_value, BaseExceptionGroup ) if is_exception_group: (_, exceptions) = exceptions_from_error( exc_type=exc_type, exc_value=exc_value, tb=tb, client_options=client_options, mechanism=mechanism, exception_id=0, parent_id=0, ) else: exceptions = [] for exc_type, exc_value, tb in walk_exception_chain(exc_info): exceptions.append( single_exception_from_error_tuple( exc_type, exc_value, tb, client_options, mechanism ) ) exceptions.reverse() return exceptions def to_string(value): # type: (str) -> str try: return text_type(value) except UnicodeDecodeError: return repr(value)[1:-1] def iter_event_stacktraces(event): # type: (Dict[str, Any]) -> Iterator[Dict[str, Any]] if "stacktrace" in event: yield event["stacktrace"] if "threads" in event: for thread in event["threads"].get("values") or (): if "stacktrace" in thread: yield thread["stacktrace"] if "exception" in event: for exception in event["exception"].get("values") or (): if "stacktrace" in exception: yield exception["stacktrace"] def iter_event_frames(event): # type: (Dict[str, Any]) -> Iterator[Dict[str, Any]] for stacktrace in iter_event_stacktraces(event): for frame in stacktrace.get("frames") or (): yield frame def handle_in_app(event, in_app_exclude=None, in_app_include=None, project_root=None): # type: (Dict[str, Any], Optional[List[str]], Optional[List[str]], Optional[str]) -> Dict[str, Any] for stacktrace in iter_event_stacktraces(event): set_in_app_in_frames( stacktrace.get("frames"), in_app_exclude=in_app_exclude, in_app_include=in_app_include, project_root=project_root, ) return event def set_in_app_in_frames(frames, in_app_exclude, in_app_include, project_root=None): # type: (Any, Optional[List[str]], Optional[List[str]], Optional[str]) -> Optional[Any] if not frames: return None for frame in frames: # if frame has already been marked as in_app, skip it current_in_app = frame.get("in_app") if current_in_app is not None: continue module = frame.get("module") # check if module in frame is in the list of modules to include if _module_in_list(module, in_app_include): frame["in_app"] = True continue # check if module in frame is in the list of modules to exclude if _module_in_list(module, in_app_exclude): frame["in_app"] = False continue # if frame has no abs_path, skip further checks abs_path = frame.get("abs_path") if abs_path is None: continue if _is_external_source(abs_path): frame["in_app"] = False continue if _is_in_project_root(abs_path, project_root): frame["in_app"] = True continue return frames def exc_info_from_error(error): # type: (Union[BaseException, ExcInfo]) -> ExcInfo if isinstance(error, tuple) and len(error) == 3: exc_type, exc_value, tb = error elif isinstance(error, BaseException): tb = getattr(error, "__traceback__", None) if tb is not None: exc_type = type(error) exc_value = error else: exc_type, exc_value, tb = sys.exc_info() if exc_value is not error: tb = None exc_value = error exc_type = type(error) else: raise ValueError("Expected Exception object to report, got %s!" % type(error)) return exc_type, exc_value, tb def event_from_exception( exc_info, # type: Union[BaseException, ExcInfo] client_options=None, # type: Optional[Dict[str, Any]] mechanism=None, # type: Optional[Dict[str, Any]] ): # type: (...) -> Tuple[Dict[str, Any], Dict[str, Any]] exc_info = exc_info_from_error(exc_info) hint = event_hint_with_exc_info(exc_info) return ( { "level": "error", "exception": { "values": exceptions_from_error_tuple( exc_info, client_options, mechanism ) }, }, hint, ) def _module_in_list(name, items): # type: (str, Optional[List[str]]) -> bool if name is None: return False if not items: return False for item in items: if item == name or name.startswith(item + "."): return True return False def _is_external_source(abs_path): # type: (str) -> bool # check if frame is in 'site-packages' or 'dist-packages' external_source = ( re.search(r"[\\/](?:dist\|site)-packages[\\/]", abs_path) is not None ) return external_source def _is_in_project_root(abs_path, project_root): # type: (str, Optional[str]) -> bool if project_root is None: return False # check if path is in the project root if abs_path.startswith(project_root): return True return False def strip_string(value, max_length=None): # type: (str, Optional[int]) -> Union[AnnotatedValue, str] if not value: return value if max_length is None: max_length = DEFAULT_MAX_VALUE_LENGTH length = len(value.encode("utf-8")) if length > max_length: return AnnotatedValue( value=value[: max_length - 3] + "...", metadata={ "len": length, "rem": [["!limit", "x", max_length - 3, max_length]], }, ) return value def _is_contextvars_broken(): # type: () -> bool """ Returns whether gevent/eventlet have patched the stdlib in a way where thread locals are now more "correct" than contextvars. """ try: import gevent # type: ignore from gevent.monkey import is_object_patched # type: ignore # Get the MAJOR and MINOR version numbers of Gevent version_tuple = tuple( [int(part) for part in re.split(r"a\|b\|rc\|\.", gevent.__version__)[:2]] ) if is_object_patched("threading", "local"): # Gevent 20.9.0 depends on Greenlet 0.4.17 which natively handles switching # context vars when greenlets are switched, so, Gevent 20.9.0+ is all fine. # Ref: https://github.com/gevent/gevent/blob/83c9e2ae5b0834b8f84233760aabe82c3ba065b4/src/gevent/monkey.py#L604-L609 # Gevent 20.5, that doesn't depend on Greenlet 0.4.17 with native support # for contextvars, is able to patch both thread locals and contextvars, in # that case, check if contextvars are effectively patched. if ( # Gevent 20.9.0+ (sys.version_info >= (3, 7) and version_tuple >= (20, 9)) # Gevent 20.5.0+ or Python < 3.7 or (is_object_patched("contextvars", "ContextVar")) ): return False return True except ImportError: pass try: from eventlet.patcher import is_monkey_patched # type: ignore if is_monkey_patched("thread"): return True except ImportError: pass return False def _make_threadlocal_contextvars(local): # type: (type) -> type class ContextVar(object): # Super-limited impl of ContextVar def __init__(self, name): # type: (str) -> None self._name = name self._local = local() def get(self, default): # type: (Any) -> Any return getattr(self._local, "value", default) def set(self, value): # type: (Any) -> None self._local.value = value return ContextVar def _get_contextvars(): # type: () -> Tuple[bool, type] """ Figure out the "right" contextvars installation to use. Returns a `contextvars.ContextVar`-like class with a limited API. See https://docs.sentry.io/platforms/python/contextvars/ for more information. """ if not _is_contextvars_broken(): # aiocontextvars is a PyPI package that ensures that the contextvars # backport (also a PyPI package) works with asyncio under Python 3.6 # # Import it if available. if sys.version_info < (3, 7): # `aiocontextvars` is absolutely required for functional # contextvars on Python 3.6. try: from aiocontextvars import ContextVar return True, ContextVar except ImportError: pass else: # On Python 3.7 contextvars are functional. try: from contextvars import ContextVar return True, ContextVar except ImportError: pass # Fall back to basic thread-local usage. from threading import local return False, _make_threadlocal_contextvars(local) HAS_REAL_CONTEXTVARS, ContextVar = _get_contextvars() CONTEXTVARS_ERROR_MESSAGE = """ With asyncio/ASGI applications, the Sentry SDK requires a functional installation of `contextvars` to avoid leaking scope/context data across requests. Please refer to https://docs.sentry.io/platforms/python/contextvars/ for more information. """ def qualname_from_function(func): # type: (Callable[..., Any]) -> Optional[str] """Return the qualified name of func. Works with regular function, lambda, partial and partialmethod.""" func_qualname = None # type: Optional[str] # Python 2 try: return "%s.%s.%s" % ( func.im_class.__module__, # type: ignore func.im_class.__name__, # type: ignore func.__name__, ) except Exception: pass prefix, suffix = "", "" if ( _PARTIALMETHOD_AVAILABLE and hasattr(func, "_partialmethod") and isinstance(func._partialmethod, partialmethod) ): prefix, suffix = "partialmethod(<function ", ">)" func = func._partialmethod.func elif isinstance(func, partial) and hasattr(func.func, "__name__"): prefix, suffix = "partial(<function ", ">)" func = func.func if hasattr(func, "__qualname__"): func_qualname = func.__qualname__ elif hasattr(func, "__name__"): # Python 2.7 has no __qualname__ func_qualname = func.__name__ # Python 3: methods, functions, classes if func_qualname is not None: if hasattr(func, "__module__"): func_qualname = func.__module__ + "." + func_qualname func_qualname = prefix + func_qualname + suffix return func_qualname def transaction_from_function(func): # type: (Callable[..., Any]) -> Optional[str] return qualname_from_function(func) disable_capture_event = ContextVar("disable_capture_event") class ServerlessTimeoutWarning(Exception): # noqa: N818 """Raised when a serverless method is about to reach its timeout.""" pass class TimeoutThread(threading.Thread): """Creates a Thread which runs (sleeps) for a time duration equal to waiting_time and raises a custom ServerlessTimeout exception. """ def __init__(self, waiting_time, configured_timeout): # type: (float, int) -> None threading.Thread.__init__(self) self.waiting_time = waiting_time self.configured_timeout = configured_timeout self._stop_event = threading.Event() def stop(self): # type: () -> None self._stop_event.set() def run(self): # type: () -> None self._stop_event.wait(self.waiting_time) if self._stop_event.is_set(): return integer_configured_timeout = int(self.configured_timeout) # Setting up the exact integer value of configured time(in seconds) if integer_configured_timeout < self.configured_timeout: integer_configured_timeout = integer_configured_timeout + 1 # Raising Exception after timeout duration is reached raise ServerlessTimeoutWarning( "WARNING : Function is expected to get timed out. Configured timeout duration = {} seconds.".format( integer_configured_timeout ) ) def to_base64(original): # type: (str) -> Optional[str] """ Convert a string to base64, via UTF-8. Returns None on invalid input. """ base64_string = None try: utf8_bytes = original.encode("UTF-8") base64_bytes = base64.b64encode(utf8_bytes) base64_string = base64_bytes.decode("UTF-8") except Exception as err: logger.warning("Unable to encode {orig} to base64:".format(orig=original), err) return base64_string def from_base64(base64_string): # type: (str) -> Optional[str] """ Convert a string from base64, via UTF-8. Returns None on invalid input. """ utf8_string = None try: only_valid_chars = BASE64_ALPHABET.match(base64_string) assert only_valid_chars base64_bytes = base64_string.encode("UTF-8") utf8_bytes = base64.b64decode(base64_bytes) utf8_string = utf8_bytes.decode("UTF-8") except Exception as err: logger.warning( "Unable to decode {b64} from base64:".format(b64=base64_string), err ) return utf8_string Components = namedtuple("Components", ["scheme", "netloc", "path", "query", "fragment"]) def sanitize_url(url, remove_authority=True, remove_query_values=True, split=False): # type: (str, bool, bool, bool) -> Union[str, Components] """ Removes the authority and query parameter values from a given URL. """ parsed_url = urlsplit(url) query_params = parse_qs(parsed_url.query, keep_blank_values=True) # strip username:password (netloc can be usr:pwd@example.com) if remove_authority: netloc_parts = parsed_url.netloc.split("@") if len(netloc_parts) > 1: netloc = "%s:%s@%s" % ( SENSITIVE_DATA_SUBSTITUTE, SENSITIVE_DATA_SUBSTITUTE, netloc_parts[-1], ) else: netloc = parsed_url.netloc else: netloc = parsed_url.netloc # strip values from query string if remove_query_values: query_string = unquote( urlencode({key: SENSITIVE_DATA_SUBSTITUTE for key in query_params}) ) else: query_string = parsed_url.query components = Components( scheme=parsed_url.scheme, netloc=netloc, query=query_string, path=parsed_url.path, fragment=parsed_url.fragment, ) if split: return components else: return urlunsplit(components) ParsedUrl = namedtuple("ParsedUrl", ["url", "query", "fragment"]) def parse_url(url, sanitize=True): # type: (str, bool) -> ParsedUrl """ Splits a URL into a url (including path), query and fragment. If sanitize is True, the query parameters will be sanitized to remove sensitive data. The autority (username and password) in the URL will always be removed. """ parsed_url = sanitize_url( url, remove_authority=True, remove_query_values=sanitize, split=True ) base_url = urlunsplit( Components( scheme=parsed_url.scheme, # type: ignore netloc=parsed_url.netloc, # type: ignore query="", path=parsed_url.path, # type: ignore fragment="", ) ) return ParsedUrl( url=base_url, query=parsed_url.query, # type: ignore fragment=parsed_url.fragment, # type: ignore ) def is_valid_sample_rate(rate, source): # type: (Any, str) -> bool """ Checks the given sample rate to make sure it is valid type and value (a boolean or a number between 0 and 1, inclusive). """ # both booleans and NaN are instances of Real, so a) checking for Real # checks for the possibility of a boolean also, and b) we have to check # separately for NaN and Decimal does not derive from Real so need to check that too if not isinstance(rate, (Real, Decimal)) or math.isnan(rate): logger.warning( "{source} Given sample rate is invalid. Sample rate must be a boolean or a number between 0 and 1. Got {rate} of type {type}.".format( source=source, rate=rate, type=type(rate) ) ) return False # in case rate is a boolean, it will get cast to 1 if it's True and 0 if it's False rate = float(rate) if rate < 0 or rate > 1: logger.warning( "{source} Given sample rate is invalid. Sample rate must be between 0 and 1. Got {rate}.".format( source=source, rate=rate ) ) return False return True def match_regex_list(item, regex_list=None, substring_matching=False): # type: (str, Optional[List[str]], bool) -> bool if regex_list is None: return False for item_matcher in regex_list: if not substring_matching and item_matcher[-1] != "$": item_matcher += "$" matched = re.search(item_matcher, item) if matched: return True return False def parse_version(version): # type: (str) -> Optional[Tuple[int, ...]] """ Parses a version string into a tuple of integers. This uses the parsing loging from PEP 440: https://peps.python.org/pep-0440/#appendix-b-parsing-version-strings-with-regular-expressions """ VERSION_PATTERN = r""" # noqa: N806 v? (?: (?:(?P<epoch>[0-9]+)!)? # epoch (?P<release>[0-9]+(?:\.[0-9]+)) # release segment (?P<pre> # pre-release [-_\.]? (?P<pre_l>(a\|b\|c\|rc\|alpha\|beta\|pre\|preview)) [-_\.]? (?P<pre_n>[0-9]+)? )? (?P<post> # post release (?:-(?P<post_n1>[0-9]+)) \| (?: [-_\.]? (?P<post_l>post\|rev\|r) [-_\.]? (?P<post_n2>[0-9]+)? ) )? (?P<dev> # dev release [-_\.]? (?P<dev_l>dev) [-_\.]? (?P<dev_n>[0-9]+)? )? ) (?:\+(?P<local>[a-z0-9]+(?:[-_\.][a-z0-9]+)))? # local version """ pattern = re.compile( r"^\s" + VERSION_PATTERN + r"\s$", re.VERBOSE \| re.IGNORECASE, ) try: release = pattern.match(version).groupdict()["release"] # type: ignore release_tuple = tuple(map(int, release.split(".")[:3])) # type: Tuple[int, ...] except (TypeError, ValueError, AttributeError): return None return release_tuple if PY37: def nanosecond_time(): # type: () -> int return time.perf_counter_ns() elif PY33: def nanosecond_time(): # type: () -> int return int(time.perf_counter() 1e9) else: def nanosecond_time(): # type: () -> int raise AttributeError if PY2: def now(): # type: () -> float return time.time() else: def now(): # type: () -> float return time.perf_counter()	45.58 KB	2026-01-20 12:58
📄 worker.py import os import threading from time import sleep, time from sentry_sdk._compat import check_thread_support from sentry_sdk._queue import Queue, FullError from sentry_sdk.utils import logger from sentry_sdk.consts import DEFAULT_QUEUE_SIZE from sentry_sdk._types import TYPE_CHECKING if TYPE_CHECKING: from typing import Any from typing import Optional from typing import Callable _TERMINATOR = object() class BackgroundWorker(object): def __init__(self, queue_size=DEFAULT_QUEUE_SIZE): # type: (int) -> None check_thread_support() self._queue = Queue(queue_size) # type: Queue self._lock = threading.Lock() self._thread = None # type: Optional[threading.Thread] self._thread_for_pid = None # type: Optional[int] @property def is_alive(self): # type: () -> bool if self._thread_for_pid != os.getpid(): return False if not self._thread: return False return self._thread.is_alive() def _ensure_thread(self): # type: () -> None if not self.is_alive: self.start() def _timed_queue_join(self, timeout): # type: (float) -> bool deadline = time() + timeout queue = self._queue queue.all_tasks_done.acquire() try: while queue.unfinished_tasks: delay = deadline - time() if delay <= 0: return False queue.all_tasks_done.wait(timeout=delay) return True finally: queue.all_tasks_done.release() def start(self): # type: () -> None with self._lock: if not self.is_alive: self._thread = threading.Thread( target=self._target, name="raven-sentry.BackgroundWorker" ) self._thread.daemon = True self._thread.start() self._thread_for_pid = os.getpid() def kill(self): # type: () -> None """ Kill worker thread. Returns immediately. Not useful for waiting on shutdown for events, use `flush` for that. """ logger.debug("background worker got kill request") with self._lock: if self._thread: try: self._queue.put_nowait(_TERMINATOR) except FullError: logger.debug("background worker queue full, kill failed") self._thread = None self._thread_for_pid = None def flush(self, timeout, callback=None): # type: (float, Optional[Any]) -> None logger.debug("background worker got flush request") with self._lock: if self.is_alive and timeout > 0.0: self._wait_flush(timeout, callback) logger.debug("background worker flushed") def full(self): # type: () -> bool return self._queue.full() def _wait_flush(self, timeout, callback): # type: (float, Optional[Any]) -> None initial_timeout = min(0.1, timeout) if not self._timed_queue_join(initial_timeout): pending = self._queue.qsize() + 1 logger.debug("%d event(s) pending on flush", pending) if callback is not None: callback(pending, timeout) if not self._timed_queue_join(timeout - initial_timeout): pending = self._queue.qsize() + 1 logger.error("flush timed out, dropped %s events", pending) def submit(self, callback): # type: (Callable[[], None]) -> bool self._ensure_thread() try: self._queue.put_nowait(callback) return True except FullError: return False def _target(self): # type: () -> None while True: callback = self._queue.get() try: if callback is _TERMINATOR: break try: callback() except Exception: logger.error("Failed processing job", exc_info=True) finally: self._queue.task_done() sleep(0)	4.15 KB	2026-01-20 12:58

Edit File:

Terminal