File Manager

X7ROOT File Manager

Current Path: /opt/hc_python/lib/python3.12/site-packages/sentry_sdk/integrations

opt / hc_python / lib / python3.12 / site-packages / sentry_sdk / integrations /

?? ..
?? __init__.py(12.51 KB)
?? __pycache__
?? _asgi_common.py(4 KB)
?? _wsgi_common.py(7.28 KB)
?? aiohttp.py(19.28 KB)
?? aiomysql.py(9.09 KB)
?? anthropic.py(39 KB)
?? argv.py(876 B)
?? ariadne.py(5.7 KB)
?? arq.py(9.23 KB)
?? asgi.py(20.06 KB)
?? asyncio.py(9.28 KB)
?? asyncpg.py(9.68 KB)
?? atexit.py(1.51 KB)
?? aws_lambda.py(17.41 KB)
?? beam.py(4.91 KB)
?? boto3.py(6.2 KB)
?? bottle.py(7.21 KB)
?? celery
?? chalice.py(4.51 KB)
?? clickhouse_driver.py(5.85 KB)
?? cloud_resource_context.py(7.49 KB)
?? cohere.py(10.44 KB)
?? dedupe.py(1.86 KB)
?? django
?? dramatiq.py(8.02 KB)
?? excepthook.py(2.25 KB)
?? executing.py(1.93 KB)
?? falcon.py(9.04 KB)
?? fastapi.py(5.28 KB)
?? flask.py(8.27 KB)
?? gcp.py(10.57 KB)
?? gnu_backtrace.py(2.72 KB)
?? google_genai
?? gql.py(4.93 KB)
?? graphene.py(5.71 KB)
?? grpc
?? httpx.py(9.79 KB)
?? httpx2.py(9.8 KB)
?? huey.py(8.19 KB)
?? huggingface_hub.py(15.28 KB)
?? langchain.py(48.31 KB)
?? langgraph.py(18.13 KB)
?? launchdarkly.py(1.87 KB)
?? litellm.py(13.03 KB)
?? litestar.py(11.46 KB)
?? logging.py(15.69 KB)
?? loguru.py(6.35 KB)
?? mcp.py(23.12 KB)
?? modules.py(787 B)
?? openai.py(53.38 KB)
?? openai_agents
?? openfeature.py(1.08 KB)
?? opentelemetry
?? otlp.py(7.99 KB)
?? pure_eval.py(4.41 KB)
?? pydantic_ai
?? pymongo.py(8.21 KB)
?? pyramid.py(7.42 KB)
?? pyreqwest.py(6.82 KB)
?? quart.py(7.32 KB)
?? ray.py(5.75 KB)
?? redis
?? rq.py(7.81 KB)
?? rust_tracing.py(9.44 KB)
?? sanic.py(15.25 KB)
?? serverless.py(1.58 KB)
?? socket.py(5.02 KB)
?? spark
?? sqlalchemy.py(5.24 KB)
?? starlette.py(27.93 KB)
?? starlite.py(11.04 KB)
?? statsig.py(1.19 KB)
?? stdlib.py(14.01 KB)
?? strawberry.py(17.39 KB)
?? sys_exit.py(2.35 KB)
?? threading.py(6.88 KB)
?? tornado.py(10.79 KB)
?? trytond.py(1.67 KB)
?? typer.py(1.72 KB)
?? unleash.py(1.02 KB)
?? unraisablehook.py(1.65 KB)
?? wsgi.py(15.03 KB)

Editing: gcp.py

import functools
import sys
from copy import deepcopy
from datetime import datetime, timedelta, timezone
from os import environ
from typing import TYPE_CHECKING

import sentry_sdk
from sentry_sdk.api import continue_trace
from sentry_sdk.consts import OP
from sentry_sdk.integrations import Integration
from sentry_sdk.integrations._wsgi_common import _filter_headers
from sentry_sdk.integrations.cloud_resource_context import CLOUD_PROVIDER
from sentry_sdk.scope import Scope, should_send_default_pii
from sentry_sdk.traces import SegmentSource
from sentry_sdk.tracing import TransactionSource
from sentry_sdk.tracing_utils import has_span_streaming_enabled
from sentry_sdk.utils import (
    AnnotatedValue,
    TimeoutThread,
    capture_internal_exceptions,
    event_from_exception,
    logger,
    reraise,
)

# Constants
TIMEOUT_WARNING_BUFFER = 1.5  # Buffer time required to send timeout warning to Sentry
MILLIS_TO_SECONDS = 1000.0

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

from sentry_sdk._types import Event, EventProcessor, Hint

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

def _wrap_func(func: "F") -> "F":
    @functools.wraps(func)
    def sentry_func(
        functionhandler: "Any", gcp_event: "Any", *args: "Any", **kwargs: "Any"
    ) -> "Any":
        client = sentry_sdk.get_client()

integration = client.get_integration(GcpIntegration)
        if integration is None:
            return func(functionhandler, gcp_event, *args, **kwargs)

configured_time = environ.get("FUNCTION_TIMEOUT_SEC")
        if not configured_time:
            logger.debug(
                "The configured timeout could not be fetched from Cloud Functions configuration."
            )
            return func(functionhandler, gcp_event, *args, **kwargs)

configured_time = int(configured_time)

initial_time = datetime.now(timezone.utc)

with sentry_sdk.isolation_scope() as scope:
            with capture_internal_exceptions():
                scope.clear_breadcrumbs()
                scope.add_event_processor(
                    _make_request_event_processor(
                        gcp_event, configured_time, initial_time
                    )
                )
                scope.set_tag("gcp_region", environ.get("FUNCTION_REGION"))
                timeout_thread = None
                if (
                    integration.timeout_warning
                    and configured_time > TIMEOUT_WARNING_BUFFER
                ):
                    waiting_time = configured_time - TIMEOUT_WARNING_BUFFER

timeout_thread = TimeoutThread(
                        waiting_time,
                        configured_time,
                        isolation_scope=scope,
                        current_scope=sentry_sdk.get_current_scope(),
                    )

# Starting the thread to raise timeout warning exception
                    timeout_thread.start()

headers = {}
            header_attributes: "dict[str, Any]" = {}
            if hasattr(gcp_event, "headers"):
                headers = gcp_event.headers
                for header, header_value in _filter_headers(
                    headers, use_annotated_value=False
                ).items():
                    header_attributes[f"http.request.header.{header.lower()}"] = (
                        # header_value will always be a string because we set `use_annotated_value` to false above
                        header_value
                    )

additional_attributes = {}
            if hasattr(gcp_event, "method"):
                additional_attributes["http.request.method"] = gcp_event.method

if should_send_default_pii() and hasattr(gcp_event, "query_string"):
                additional_attributes["url.query"] = gcp_event.query_string.decode(
                    "utf-8", errors="replace"
                )

sampling_context = {
                "gcp_env": {
                    "function_name": environ.get("FUNCTION_NAME"),
                    "function_entry_point": environ.get("ENTRY_POINT"),
                    "function_identity": environ.get("FUNCTION_IDENTITY"),
                    "function_region": environ.get("FUNCTION_REGION"),
                    "function_project": environ.get("GCP_PROJECT"),
                },
                "gcp_event": gcp_event,
            }

function_name = environ.get("FUNCTION_NAME", "<unknown GCP function>")

if environ.get("GCP_PROJECT"):
                additional_attributes["gcp.project.id"] = environ.get("GCP_PROJECT")

if environ.get("FUNCTION_IDENTITY"):
                additional_attributes["faas.identity"] = environ.get(
                    "FUNCTION_IDENTITY"
                )

if environ.get("ENTRY_POINT"):
                additional_attributes["faas.entry_point"] = environ.get("ENTRY_POINT")

if has_span_streaming_enabled(client.options):
                sentry_sdk.traces.continue_trace(headers)
                Scope.set_custom_sampling_context(sampling_context)
                span_ctx = sentry_sdk.traces.start_span(
                    name=function_name,
                    parent_span=None,
                    attributes={
                        "sentry.op": OP.FUNCTION_GCP,
                        "sentry.origin": GcpIntegration.origin,
                        "sentry.span.source": SegmentSource.COMPONENT,
                        "cloud.provider": CLOUD_PROVIDER.GCP,
                        "faas.name": function_name,
                        **header_attributes,
                        **additional_attributes,
                    },
                )
            else:
                transaction = continue_trace(
                    headers,
                    op=OP.FUNCTION_GCP,
                    name=environ.get("FUNCTION_NAME", ""),
                    source=TransactionSource.COMPONENT,
                    origin=GcpIntegration.origin,
                )

span_ctx = sentry_sdk.start_transaction(
                    transaction, custom_sampling_context=sampling_context
                )

with span_ctx:
                try:
                    return func(functionhandler, gcp_event, *args, **kwargs)
                except Exception:
                    exc_info = sys.exc_info()
                    sentry_event, hint = event_from_exception(
                        exc_info,
                        client_options=client.options,
                        mechanism={"type": "gcp", "handled": False},
                    )
                    sentry_sdk.capture_event(sentry_event, hint=hint)
                    reraise(*exc_info)
                finally:
                    if timeout_thread:
                        timeout_thread.stop()
                    # Flush out the event queue
                    client.flush()

return sentry_func  # type: ignore

class GcpIntegration(Integration):
    identifier = "gcp"
    origin = f"auto.function.{identifier}"

def __init__(self, timeout_warning: bool = False) -> None:
        self.timeout_warning = timeout_warning

@staticmethod
    def setup_once() -> None:
        import __main__ as gcp_functions

if not hasattr(gcp_functions, "worker_v1"):
            logger.warning(
                "GcpIntegration currently supports only Python 3.7 runtime environment."
            )
            return

worker1 = gcp_functions.worker_v1

worker1.FunctionHandler.invoke_user_function = _wrap_func(
            worker1.FunctionHandler.invoke_user_function
        )

def _make_request_event_processor(
    gcp_event: "Any", configured_timeout: "Any", initial_time: "Any"
) -> "EventProcessor":
    def event_processor(event: "Event", hint: "Hint") -> "Optional[Event]":
        final_time = datetime.now(timezone.utc)
        time_diff = final_time - initial_time

execution_duration_in_millis = time_diff / timedelta(milliseconds=1)

extra = event.setdefault("extra", {})
        extra["google cloud functions"] = {
            "function_name": environ.get("FUNCTION_NAME"),
            "function_entry_point": environ.get("ENTRY_POINT"),
            "function_identity": environ.get("FUNCTION_IDENTITY"),
            "function_region": environ.get("FUNCTION_REGION"),
            "function_project": environ.get("GCP_PROJECT"),
            "execution_duration_in_millis": execution_duration_in_millis,
            "configured_timeout_in_seconds": configured_timeout,
        }

extra["google cloud logs"] = {
            "url": _get_google_cloud_logs_url(final_time),
        }

request = event.get("request", {})

request["url"] = "gcp:///{}".format(environ.get("FUNCTION_NAME"))

if hasattr(gcp_event, "method"):
            request["method"] = gcp_event.method

if hasattr(gcp_event, "query_string"):
            request["query_string"] = gcp_event.query_string.decode(
                "utf-8", errors="replace"
            )

if hasattr(gcp_event, "headers"):
            request["headers"] = _filter_headers(gcp_event.headers)

if should_send_default_pii():
            if hasattr(gcp_event, "data"):
                request["data"] = gcp_event.data
        else:
            if hasattr(gcp_event, "data"):
                # Unfortunately couldn't find a way to get structured body from GCP
                # event. Meaning every body is unstructured to us.
                request["data"] = AnnotatedValue.removed_because_raw_data()

event["request"] = deepcopy(request)

return event

return event_processor

def _get_google_cloud_logs_url(final_time: "datetime") -> str:
    """
    Generates a Google Cloud Logs console URL based on the environment variables
    Arguments:
        final_time {datetime} -- Final time
    Returns:
        str -- Google Cloud Logs Console URL to logs.
    """
    hour_ago = final_time - timedelta(hours=1)
    formatstring = "%Y-%m-%dT%H:%M:%SZ"

url = (
        "https://console.cloud.google.com/logs/viewer?project={project}&resource=cloud_function"
        "%2Ffunction_name%2F{function_name}%2Fregion%2F{region}&minLogLevel=0&expandAll=false"
        "&timestamp={timestamp_end}&customFacets=&limitCustomFacetWidth=true"
        "&dateRangeStart={timestamp_start}&dateRangeEnd={timestamp_end}"
        "&interval=PT1H&scrollTimestamp={timestamp_end}"
    ).format(
        project=environ.get("GCP_PROJECT"),
        function_name=environ.get("FUNCTION_NAME"),
        region=environ.get("FUNCTION_REGION"),
        timestamp_end=final_time.strftime(formatstring),
        timestamp_start=hour_ago.strftime(formatstring),
    )

return url

X7ROOT File Manager

Editing: gcp.py

Upload File

Create Folder