File Manager

X7ROOT File Manager

Current Path: /opt/cloudlinux/venv/lib/python3.11/site-packages/sentry_sdk/integrations

opt / cloudlinux / venv / lib / python3.11 / site-packages / sentry_sdk / integrations /

?? ..
?? __init__.py(6.67 KB)
?? __pycache__
?? _wsgi_common.py(4.43 KB)
?? aiohttp.py(11.28 KB)
?? argv.py(963 B)
?? arq.py(6.58 KB)
?? asgi.py(11.54 KB)
?? asyncio.py(2.98 KB)
?? atexit.py(1.8 KB)
?? aws_lambda.py(15.44 KB)
?? beam.py(5.56 KB)
?? boto3.py(4.44 KB)
?? bottle.py(6.32 KB)
?? celery.py(18.65 KB)
?? chalice.py(4.66 KB)
?? cloud_resource_context.py(6.6 KB)
?? dedupe.py(1.16 KB)
?? django
?? excepthook.py(2.21 KB)
?? executing.py(1.99 KB)
?? falcon.py(7.8 KB)
?? fastapi.py(4.39 KB)
?? flask.py(7.72 KB)
?? gcp.py(8.02 KB)
?? gnu_backtrace.py(2.86 KB)
?? grpc
?? httpx.py(4.89 KB)
?? huey.py(4.59 KB)
?? logging.py(8.97 KB)
?? loguru.py(2.98 KB)
?? modules.py(2.06 KB)
?? opentelemetry
?? pure_eval.py(4.45 KB)
?? pymongo.py(5.87 KB)
?? pyramid.py(7.27 KB)
?? quart.py(7.2 KB)
?? redis
?? rq.py(5.28 KB)
?? sanic.py(11.06 KB)
?? serverless.py(1.93 KB)
?? socket.py(2.88 KB)
?? spark
?? sqlalchemy.py(4.14 KB)
?? starlette.py(22.67 KB)
?? starlite.py(9.85 KB)
?? stdlib.py(8.06 KB)
?? threading.py(2.87 KB)
?? tornado.py(7.17 KB)
?? trytond.py(1.7 KB)
?? wsgi.py(9.36 KB)

Editing: huey.py

from __future__ import absolute_import

import sys
from datetime import datetime

from sentry_sdk._compat import reraise
from sentry_sdk._types import TYPE_CHECKING
from sentry_sdk import Hub
from sentry_sdk.consts import OP
from sentry_sdk.hub import _should_send_default_pii
from sentry_sdk.integrations import DidNotEnable, Integration
from sentry_sdk.tracing import Transaction, TRANSACTION_SOURCE_TASK
from sentry_sdk.utils import (
    capture_internal_exceptions,
    event_from_exception,
    SENSITIVE_DATA_SUBSTITUTE,
)

if TYPE_CHECKING:
    from typing import Any, Callable, Optional, Union, TypeVar

from sentry_sdk._types import EventProcessor, Event, Hint
    from sentry_sdk.utils import ExcInfo

F = TypeVar("F", bound=Callable[..., Any])

try:
    from huey.api import Huey, Result, ResultGroup, Task
    from huey.exceptions import CancelExecution, RetryTask, TaskLockedException
except ImportError:
    raise DidNotEnable("Huey is not installed")

HUEY_CONTROL_FLOW_EXCEPTIONS = (CancelExecution, RetryTask, TaskLockedException)

class HueyIntegration(Integration):
    identifier = "huey"

@staticmethod
    def setup_once():
        # type: () -> None
        patch_enqueue()
        patch_execute()

def patch_enqueue():
    # type: () -> None
    old_enqueue = Huey.enqueue

def _sentry_enqueue(self, task):
        # type: (Huey, Task) -> Optional[Union[Result, ResultGroup]]
        hub = Hub.current

if hub.get_integration(HueyIntegration) is None:
            return old_enqueue(self, task)

with hub.start_span(op=OP.QUEUE_SUBMIT_HUEY, description=task.name):
            return old_enqueue(self, task)

Huey.enqueue = _sentry_enqueue

def _make_event_processor(task):
    # type: (Any) -> EventProcessor
    def event_processor(event, hint):
        # type: (Event, Hint) -> Optional[Event]

with capture_internal_exceptions():
            tags = event.setdefault("tags", {})
            tags["huey_task_id"] = task.id
            tags["huey_task_retry"] = task.default_retries > task.retries
            extra = event.setdefault("extra", {})
            extra["huey-job"] = {
                "task": task.name,
                "args": task.args
                if _should_send_default_pii()
                else SENSITIVE_DATA_SUBSTITUTE,
                "kwargs": task.kwargs
                if _should_send_default_pii()
                else SENSITIVE_DATA_SUBSTITUTE,
                "retry": (task.default_retries or 0) - task.retries,
            }

return event

return event_processor

def _capture_exception(exc_info):
    # type: (ExcInfo) -> None
    hub = Hub.current

if exc_info[0] in HUEY_CONTROL_FLOW_EXCEPTIONS:
        hub.scope.transaction.set_status("aborted")
        return

hub.scope.transaction.set_status("internal_error")
    event, hint = event_from_exception(
        exc_info,
        client_options=hub.client.options if hub.client else None,
        mechanism={"type": HueyIntegration.identifier, "handled": False},
    )
    hub.capture_event(event, hint=hint)

def _wrap_task_execute(func):
    # type: (F) -> F
    def _sentry_execute(*args, **kwargs):
        # type: (*Any, **Any) -> Any
        hub = Hub.current
        if hub.get_integration(HueyIntegration) is None:
            return func(*args, **kwargs)

try:
            result = func(*args, **kwargs)
        except Exception:
            exc_info = sys.exc_info()
            _capture_exception(exc_info)
            reraise(*exc_info)

return result

return _sentry_execute  # type: ignore

def patch_execute():
    # type: () -> None
    old_execute = Huey._execute

def _sentry_execute(self, task, timestamp=None):
        # type: (Huey, Task, Optional[datetime]) -> Any
        hub = Hub.current

if hub.get_integration(HueyIntegration) is None:
            return old_execute(self, task, timestamp)

with hub.push_scope() as scope:
            with capture_internal_exceptions():
                scope._name = "huey"
                scope.clear_breadcrumbs()
                scope.add_event_processor(_make_event_processor(task))

transaction = Transaction(
                name=task.name,
                status="ok",
                op=OP.QUEUE_TASK_HUEY,
                source=TRANSACTION_SOURCE_TASK,
            )

if not getattr(task, "_sentry_is_patched", False):
                task.execute = _wrap_task_execute(task.execute)
                task._sentry_is_patched = True

with hub.start_transaction(transaction):
                return old_execute(self, task, timestamp)

Huey._execute = _sentry_execute

X7ROOT File Manager

Editing: huey.py

Upload File

Create Folder