class BreadcrumbType(Enum):
class Breadcrumb:
class Breadcrumbs:
    NAVIGATION = 'navigation'
    REQUEST = 'request'
    PROCESS = 'process'
    LOG = 'log'
    USER = 'user'
    STATE = 'state'
    ERROR = 'error'
    MANUAL = 'manual'
    def __init__(
        self,
        message: str,
        type: BreadcrumbType,
        metadata: Dict[str, Any],
        timestamp: str
    ):
    def timestamp(self) -> str:
    def to_dict(self) -> Dict[str, Union[str, FilterDict]]:
    def __init__(self, max_breadcrumbs: int):
    def append(self, breadcrumb: Breadcrumb) -> None:
    def resize(self, new_max: int) -> None:
    def create_copy_for_context(self) -> None:
    def clear(self) -> None:
    def to_list(self) -> List[Breadcrumb]:
    def _breadcrumbs(self): class SessionMiddleware:
    """
    Session middleware ensures that a session is appended to the event.
    """
    def __init__(self, bugsnag: Callable[[Event], Callable]):
        self.bugsnag = bugsnag

    def __call__(self, event: Event):
        session = _session_info.get()
        if session:
            if event.unhandled:
                session['events']['unhandled'] += 1
            else:
                session['events']['handled'] += 1
            event.session = deepcopy(session)
        self.bugsnag(event) class DefaultMiddleware:
    """
    DefaultMiddleware provides the transformation from request_config into
    metadata that has always been supported by bugsnag-python.
    """
    def __init__(self, bugsnag: Middleware):
        self.bugsnag = bugsnag

    def __call__(self, event: Event):
        config = event.request_config
        event.set_user(id=config.user_id)
        event.set_user(**config.user)

        if not event.context:
            event.context = config.get("context")

        for name, dictionary in config.metadata.items():
            if name in event.metadata:
                for key, value in dictionary.items():
                    if key not in event.metadata[name]:
                        event.metadata[name][key] = value
            else:
                event.add_tab(name, dictionary)

        event.add_tab("request", config.get("request_data"))
        if event.config.send_environment:
            event.add_tab("environment", config.get("environment_data"))
        event.add_tab("session", config.get("session_data"))
        event.add_tab("extraData", config.get("extra_data"))

        self.bugsnag(event) class MiddlewareStack:
    """
    Manages a stack of Bugsnag middleware.
    """
    def __init__(self):
        self.stack = []

    def before_notify(self, func: Middleware):
        """
        Add a function to be run before bugsnag is notified.

        This lets you modify the payload that will be sent.
        If your function returns False, nothing will be sent.

        >>> def add_request_data(event):
        ...    event.add_tab("request", request_data)
        >>>
        >>> stack = MiddlewareStack()
        >>> stack.before_notify(add_request_data)
        """
        self.append(SimpleMiddleware(before=func))

    def after_notify(self, func: Middleware):
        """
        Add a function to be run after bugsnag is notified.

        This lets you log errors in custom ways.
        """
        self.append(SimpleMiddleware(after=func))

    def append(self, middleware: Middleware):
        """
        Add a middleware to the end of the stack.

        It will be run after all middleware currently defined.
        If you want to stop the event progress, return from
        your __call__ method without calling the next level.

        >>> class ExampleMiddleware():
        ...     def __init__(self, bugsnag):
        ...         self.bugsnag = bugsnag
        ...     def __call__(self, event):
        ...         config = event.request_config
        ...         event.add_tab("request", config.get("request"))
        ...         self.bugsnag(event)
        ...
        >>> stack = MiddlewareStack()
        >>> stack.append(ExampleMiddleware)
        """
        self.stack.append(middleware)

    def insert_before(self, target_class: Type, middleware: Middleware):
        """
        Adds a middleware to the stack in the position before
        the target_class.
        """
        try:
            index = self.stack.index(target_class)
            self.stack.insert(index, middleware)
        except ValueError:
            self.append(middleware)

    def insert_after(self, target_class: Type, middleware: Middleware):
        """
        Adds a middleware to the stack in the position after
        the target_class.
        """
        try:
            index = self.stack.index(target_class)
            self.stack.insert(index + 1, middleware)
        except ValueError:
            self.append(middleware)

    def run(self, event: Event, callback: Callable[[], None]):
        """
        Run all the middleware in order, then call the callback.
        """

        # the last step in the event stack is to call the callback.
        # we also do this inside the exception handler, so need to ensure that
        # the callback is only called once.
        def finish(event):
            if not hasattr(finish, 'called'):
                finish.called = True
                callback()

        to_call = finish
        for middleware in reversed(self.stack):
            to_call = middleware(to_call)

        try:
            to_call(event)
        except Exception:
            event.config.logger.exception(
                'Error in exception middleware'
            )

            # still notify if middleware crashes before event
            finish(event) MAX_PAYLOAD_LENGTH = 128 * 1024
MAX_STRING_LENGTH = 1024
    LOCALS = None
class SanitizingJSONEncoder(JSONEncoder):
class FilterDict(dict):
class ThreadContextVar:
    def __init__(self, logger: logging.Logger, keyword_filters=None, **kwargs):
    def encode(self, obj):
    def filter_string_values(self, obj, ignored=None, seen=None):
    def default(self, obj):
    def _sanitize(self, obj, trim_strings, ignored=None, seen=None):
    def _sanitize_dict_key_value(self, clean_dict, key, clean_value):
    def _sanitize_dict(self, obj, trim_strings, ignored, seen):
    def _should_filter(self, key):
def parse_content_type(value: str) -> ContentType:
def is_json_content_type(value: str) -> bool:
def fully_qualified_class_name(obj):
def package_version(package_name):
def _validate_setter(types, func, should_error=False):
    def wrapper(obj, value):
    def local_context(cls):
    def __init__(self, name, **kwargs):
    def get(self):
    def set(self, new_value):
def sanitize_url(url_to_sanitize: AnyStr) -> Optional[AnyStr]:
    def to_rfc3339(dt: datetime) -> str:
    def _get_timezone_offset(dt: datetime) -> str:
    def to_rfc3339(dt: datetime) -> str: def create_default_delivery():
    if requests is not None:
        return RequestsDelivery()

    return UrllibDelivery() DEFAULT_ENDPOINT = 'https://notify.bugsnag.com' DEFAULT_SESSIONS_ENDPOINT = 'https://sessions.bugsnag.com' def warn_if_running_uwsgi_without_threads():
    if running_uwsgi and not threads_enabled:
        warnings.warn(('Bugsnag cannot run asynchronously under uWSGI' +
                       ' without enabling thread support. Please run uwsgi' +
                       ' with --enable-threads'), RuntimeWarning) class Error:
    __slots__ = (
        'error_class',
        'error_message',
        'stacktrace',
        'type',
    )

    def __init__(
        self,
        error_class: str,
        error_message: str,
        stacktrace: List[Dict[str, Any]]
    ):
        self.error_class = error_class
        self.error_message = error_message
        self.stacktrace = stacktrace
        self.type = 'python'

    def to_dict(self) -> Dict[str, Any]:
        return {
            'errorClass': self.error_class,
            'message': self.error_message,
            'stacktrace': self.stacktrace,
            'type': self.type,
        }
import os
import platform
import socket
import sys
import sysconfig
import warnings
import logging
from typing import List, Any, Tuple, Union, Optional
from threading import Lock
from bugsnag.breadcrumbs import (
    BreadcrumbType,
    Breadcrumb,
    Breadcrumbs,
    OnBreadcrumbCallback
)
from bugsnag.sessiontracker import SessionMiddleware
from bugsnag.middleware import DefaultMiddleware, MiddlewareStack
from bugsnag.utils import (fully_qualified_class_name, validate_str_setter,
                           validate_bool_setter, validate_iterable_setter,
                           validate_required_str_setter, validate_int_setter)
from bugsnag.delivery import (create_default_delivery, DEFAULT_ENDPOINT,
                              DEFAULT_SESSIONS_ENDPOINT)
from bugsnag.uwsgi import warn_if_running_uwsgi_without_threads
from bugsnag.error import Error
    from contextvars import ContextVar
    from bugsnag.utils import ThreadContextVar
        self.project_root = os.getcwd()
        self.app_type = None
        self.app_version = None
        self.params_filters = ["password", "password_confirmation", "cookie",
                               "authorization"]
        self.ignore_classes = [
            "KeyboardInterrupt",
            "django.http.Http404",
            "django.http.response.Http404",
        ]
        self.endpoint = DEFAULT_ENDPOINT
        self.session_endpoint = DEFAULT_SESSIONS_ENDPOINT
        self.auto_capture_sessions = True
        self.traceback_exclude_modules = []

        self.middleware = MiddlewareStack()

        self.internal_middleware = MiddlewareStack()
        self.internal_middleware.append(DefaultMiddleware)
        self.internal_middleware.append(SessionMiddleware)

        self.proxy_host = None

        if not os.getenv("DYNO"):
            self.hostname = socket.gethostname()
        else:
            self.hostname = None

        self.runtime_versions = {"python": platform.python_version()}

        self.logger = logger

        self._max_breadcrumbs = 25
        self.breadcrumb_log_level = logging.INFO
        self.enabled_breadcrumb_types = list(BreadcrumbType)
        self._breadcrumbs = Breadcrumbs(self.max_breadcrumbs)
        self._on_breadcrumbs = []

    def configure(self, api_key=None, app_type=None, app_version=None,
                  asynchronous=None, auto_notify=None,
                  auto_capture_sessions=None, delivery=None, endpoint=None,
                  hostname=None, ignore_classes=None, lib_root=None,
                  notify_release_stages=None, params_filters=None,
                  project_root=None, proxy_host=None, release_stage=None,
                  send_code=None, send_environment=None, session_endpoint=None,
                  traceback_exclude_modules=None, logger=_sentinel,
                  breadcrumb_log_level=None, enabled_breadcrumb_types=None,
                  max_breadcrumbs=None):
        """
        Validate and set configuration options. Will warn if an option is of an
        incorrect type.
        """
        if api_key is not None:
            self.api_key = api_key
        if app_type is not None:
            self.app_type = app_type
        if app_version is not None:
            self.app_version = app_version
        if asynchronous is not None:
            self.asynchronous = asynchronous
        if auto_notify is not None:
            self.auto_notify = auto_notify
        if auto_capture_sessions is not None:
            self.auto_capture_sessions = auto_capture_sessions
        if delivery is not None:
            self.delivery = delivery
        if endpoint is not None:
            self.endpoint = endpoint
        if hostname is not None:
            self.hostname = hostname
        if ignore_classes is not None:
            self.ignore_classes = ignore_classes
        if lib_root is not None:
            self.lib_root = lib_root
        if notify_release_stages is not None:
            self.notify_release_stages = notify_release_stages
        if params_filters is not None:
            self.params_filters = params_filters
        if project_root is not None:
            self.project_root = project_root
        if proxy_host is not None:
            self.proxy_host = proxy_host
        if release_stage is not None:
            self.release_stage = release_stage
        if send_code is not None:
            self.send_code = send_code
        if send_environment is not None:
            self.send_environment = send_environment
        if session_endpoint is not None:
            self.session_endpoint = session_endpoint
        if traceback_exclude_modules is not None:
            self.traceback_exclude_modules = traceback_exclude_modules
        if logger is not _sentinel:
            self.logger = logger
        if breadcrumb_log_level is not None:
            self.breadcrumb_log_level = breadcrumb_log_level
        if enabled_breadcrumb_types is not None:
            self.enabled_breadcrumb_types = enabled_breadcrumb_types
        if max_breadcrumbs is not None:
            self.max_breadcrumbs = max_breadcrumbs

        return self

    def get(self, name):
        """
        Get a single configuration option
        """
        warnings.warn('Using get() to retrieve a Configuration property is ' +
                      'deprecated in favor of referencing properties directly',
                      DeprecationWarning)
        return getattr(self, name)

    @property
    def api_key(self):
        """
        Unique application identifier
        """
        return self._api_key

    @api_key.setter  # type: ignore

    @validate_required_str_setter