13416015178

import atexit
import importlib.util
import logging
from contextlib import contextmanager
from contextvars import ContextVar
from datetime import datetime, timezone
from functools import partial
from os import getenv
from typing import List, Optional
from uuid import uuid4

from .api.client import AsyncClient, Client
from .api.defaults import DEFAULT_API_URL, DEFAULT_APPLICATION_ID
from .api.tracer import Tracer
from .api.types import ImpactRequest, ImpactResponse, ImpactRow, Scope3AIContext
from .constants import PROVIDERS
from .worker import BackgroundWorker

logger = logging.getLogger("scope3ai.lib")


def init_anthropic_instrumentor() -> None:
    if importlib.util.find_spec("anthropic") is not None:
        from scope3ai.tracers.anthropic.instrument import AnthropicInstrumentor

        instrumentor = AnthropicInstrumentor()
        instrumentor.instrument()


def init_cohere_instrumentor() -> None:
    if importlib.util.find_spec("cohere") is not None:
        from scope3ai.tracers.cohere.instrument import CohereInstrumentor

        instrumentor = CohereInstrumentor()
        instrumentor.instrument()


def init_openai_instrumentor() -> None:
    if importlib.util.find_spec("openai") is not None:
        from scope3ai.tracers.openai.instrument import OpenAIInstrumentor

        instrumentor = OpenAIInstrumentor()
        instrumentor.instrument()


def init_huggingface_hub_instrumentor() -> None:
    if importlib.util.find_spec("huggingface_hub") is not None:
        from scope3ai.tracers.huggingface.instrument import HuggingfaceInstrumentor

        instrumentor = HuggingfaceInstrumentor()
        instrumentor.instrument()


def init_google_genai_instrumentor() -> None:
    if importlib.util.find_spec("google") is not None:
        from scope3ai.tracers.google_genai.instrument import GoogleGenAiInstrumentor

        instrumentor = GoogleGenAiInstrumentor()
        instrumentor.instrument()


def init_litellm_instrumentor() -> None:
    if importlib.util.find_spec("litellm") is not None:
        from scope3ai.tracers.litellm.instrument import LiteLLMInstrumentor

        instrumentor = LiteLLMInstrumentor()
        instrumentor.instrument()


def init_mistral_v1_instrumentor() -> None:
    if importlib.util.find_spec("mistralai") is not None:
        from scope3ai.tracers.mistralai.instrument import MistralAIInstrumentor

        instrumentor = MistralAIInstrumentor()
        instrumentor.instrument()


def init_response_instrumentor() -> None:
    from scope3ai.response_interceptor.instrument import ResponseInterceptor

    instrumentor = ResponseInterceptor()
    instrumentor.instrument()


_INSTRUMENTS = {
    PROVIDERS.ANTROPIC.value: init_anthropic_instrumentor,
    PROVIDERS.COHERE.value: init_cohere_instrumentor,
    PROVIDERS.OPENAI.value: init_openai_instrumentor,
    PROVIDERS.HUGGINGFACE_HUB.value: init_huggingface_hub_instrumentor,
    PROVIDERS.GOOGLE_GENAI.value: init_google_genai_instrumentor,
    PROVIDERS.LITELLM.value: init_litellm_instrumentor,
    PROVIDERS.MISTRALAI.value: init_mistral_v1_instrumentor,
    PROVIDERS.RESPONSE.value: init_response_instrumentor,
}

_RE_INIT_PROVIDERS = [PROVIDERS.RESPONSE.value]


def generate_id() -> str:
    return uuid4().hex


class Scope3AIError(Exception):
    pass


class Scope3AI:
    """
    Scope3AI tracer class

    This class is a singleton that provides a context manager for tracing
    inference metadata and submitting impact requests to the Scope3 AI API.
    """

    _instance: Optional["Scope3AI"] = None
    _tracer: ContextVar[List[Tracer]] = ContextVar("tracer", default=[])
    _worker: Optional[BackgroundWorker] = None
    _providers: List[str] = []
    _keep_tracers: bool = False

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(Scope3AI, cls).__new__(cls)
        return cls._instance

    def __init__(self):
        self.api_key: Optional[str] = None
        self.api_url: Optional[str] = None
        self.sync_mode: bool = False
        self._sync_client: Optional[Client] = None
        self._async_client: Optional[AsyncClient] = None
        self.environment: Optional[str] = None
        self.client_id: Optional[str] = None
        self.project_id: Optional[str] = None
        self.application_id: Optional[str] = None

    @classmethod
    def init(
        cls,
        api_key: Optional[str] = None,
        api_url: Optional[str] = None,
        sync_mode: bool = False,
        enable_debug_logging: bool = False,
        providers: Optional[List[str]] = None,
        # metadata for scope3
        environment: Optional[str] = None,
        client_id: Optional[str] = None,
        project_id: Optional[str] = None,
        application_id: Optional[str] = None,
    ) -> "Scope3AI":
        """
        Initialize the Scope3AI SDK with the provided configuration settings.

        Args:
            api_key (str, optional): The Scope3AI API key. Can be set via `SCOPE3AI_API_KEY`
                environment variable. Required for authentication.
            api_url (str, optional): The base URL for the Scope3AI API. Can be set via
                `SCOPE3AI_API_URL` environment variable. Defaults to standard API URL.
            sync_mode (bool, optional): If True, the SDK will operate synchronously. Can be
                set via `SCOPE3AI_SYNC_MODE` environment variable. Defaults to False.
            enable_debug_logging (bool, optional): Enable debug level logging. Can be set via
                `SCOPE3AI_DEBUG_LOGGING` environment variable. Defaults to False.
            providers (List[str], optional): List of providers to instrument. If None,
                all available providers will be instrumented.
            environment (str, optional): The environment name (e.g. "production", "staging").
                Can be set via `SCOPE3AI_ENVIRONMENT` environment variable.
            client_id (str, optional): Client identifier for grouping traces. Can be set via
                `SCOPE3AI_CLIENT_ID` environment variable.
            project_id (str, optional): Project identifier for grouping traces. Can be set via
                `SCOPE3AI_PROJECT_ID` environment variable.
            application_id (str, optional): Application identifier. Can be set via
                `SCOPE3AI_APPLICATION_ID` environment variable. Defaults to "default".

        Returns:
            Scope3AI: The initialized Scope3AI instance.

        Raises:
            Scope3AIError: If the instance is already initialized or if required settings are missing.
        """
        if cls._instance is not None:
            raise Scope3AIError("Scope3AI is already initialized")
        cls._instance = self = Scope3AI()
        self.api_key = api_key or getenv("SCOPE3AI_API_KEY")
        self.api_url = api_url or getenv("SCOPE3AI_API_URL") or DEFAULT_API_URL
        self.sync_mode = sync_mode or bool(getenv("SCOPE3AI_SYNC_MODE", False))
        if not self.api_key:
            raise Scope3AIError(
                "The scope3 api_key option must be set either by "
                "passing the API key to the Scope3AI.init(api_key='xxx') "
                "or by setting the SCOPE3AI_API_KEY environment variable"
            )
        if not self.api_url:
            raise Scope3AIError(
                "The api_url option must be set either by "
                "passing the API URL to the Scope3AI.init(api_url='xxx') "
                "or by setting the SCOPE3AI_API_URL environment variable"
            )

        # metadata
        self.environment = environment or getenv("SCOPE3AI_ENVIRONMENT")
        self.client_id = client_id or getenv("SCOPE3AI_CLIENT_ID")
        self.project_id = project_id or getenv("SCOPE3AI_PROJECT_ID")
        self.application_id = (
            application_id
            or getenv("SCOPE3AI_APPLICATION_ID")
            or DEFAULT_APPLICATION_ID
        )

        if enable_debug_logging:
            self._init_logging()

        if providers is None:
            providers = list(_INSTRUMENTS.keys())

        http_client_options = {"api_key": self.api_key, "api_url": self.api_url}
        self._sync_client = Client(**http_client_options)
        self._async_client = AsyncClient(**http_client_options)
        self._init_providers(providers)
        self._init_atexit()
        return cls._instance

    @classmethod
    def get_instance(cls) -> "Scope3AI":
        """
        Return the instance of the Scope3AI singleton.

        This method provides access to the default global state of the
        Scope3AI library. The returned instance can be used to trace
        inference metadata and submit impact requests to the Scope3 AI
        API from anywhere in the application.

        Returns:
            Scope3AI: The singleton instance of the Scope3AI class.
        """
        if not cls._instance:
            raise Scope3AIError("Scope3AI is not initialized. Use Scope3AI.init()")
        return cls._instance

    def submit_impact(
        self,
        impact_row: ImpactRow,
    ) -> Scope3AIContext:
        """
        Submit an impact request to the Scope3 AI API.

        This function sends an impact request represented by the `impact_row`
        to the Scope3 AI API and optionally returns the response.

        Args:
            impact_row (ImpactRow): The impact request data
                that needs to be submitted to the Scope3 AI API.

        Returns:
            Scope3AIContext: A context object containing the request data and
            the response from the API.
        """

        def submit_impact(
            impact_row: ImpactRow,
            ctx: Scope3AIContext,
        ) -> Optional[ImpactResponse]:
            assert self._sync_client is not None
            response = self._sync_client.get_impact(
                content=ImpactRequest(rows=[impact_row]),
                with_response=True,
            )
            ctx.set_impact(response.rows[0])
            if ctx._tracer:
                ctx._tracer._unlink_trace(ctx)
            return response

        tracer = self.current_tracer
        self._fill_impact_row(impact_row, tracer, self.root_tracer)
        ctx = Scope3AIContext(request=impact_row)
        ctx._tracer = tracer
        if tracer:
            tracer._link_trace(ctx)

        if self.sync_mode:
            submit_impact(impact_row, ctx=ctx)
            return ctx

        self._ensure_worker()
        assert self._worker is not None
        self._worker.submit(partial(submit_impact, impact_row=impact_row, ctx=ctx))
        return ctx

    async def asubmit_impact(
        self,
        impact_row: ImpactRow,
    ) -> Scope3AIContext:
        """
        Async version of Scope3AI::submit_impact.
        """

        if not self.sync_mode:
            # in non sync-mode, it uses the background worker,
            # and the background worker is not async (does not have to be).
            # so we just redirect the call to the sync version.
            return self.submit_impact(impact_row)

        tracer = self.current_tracer
        self._fill_impact_row(impact_row, tracer, self.root_tracer)
        ctx = Scope3AIContext(request=impact_row)
        ctx._tracer = tracer
        if tracer:
            tracer._link_trace(ctx)

        assert self._async_client is not None
        response = await self._async_client.get_impact(
            content=ImpactRequest(rows=[impact_row]),
            with_response=True,
        )
        ctx.set_impact(response.rows[0])
        if tracer:
            tracer._unlink_trace(ctx)

        return ctx

    @property
    def root_tracer(self):
        """
        Return the root tracer.

        The root tracer is the first tracer in the current execution context
        (tracer stack). If no tracers are currently active, it returns None.

        Returns:
            Tracer: The root tracer if available, otherwise None.
        """
        tracers = self._tracer.get()
        return tracers[0] if tracers else None

    @property
    def current_tracer(self):
        """
        Return the current tracer.

        The current tracer is the last tracer in the current execution context
        (tracer stack). If no tracers are currently active, it returns None.

        Returns:
            Tracer: The current tracer if available, otherwise None.
        """
        tracers = self._tracer.get()
        return tracers[-1] if tracers else None

    @contextmanager
    def trace(
        self,
        keep_traces=False,
        client_id: Optional[str] = None,
        project_id: Optional[str] = None,
        application_id: Optional[str] = None,
        session_id: Optional[str] = None,
    ):
        root_tracer = self.root_tracer
        if not client_id:
            client_id = root_tracer.client_id if root_tracer else self.client_id
        if not project_id:
            project_id = root_tracer.project_id if root_tracer else self.project_id
        if not application_id:
            application_id = (
                root_tracer.application_id if root_tracer else self.application_id
            )
        if not session_id:
            session_id = root_tracer.session_id if root_tracer else None
        tracer = Tracer(
            keep_traces=keep_traces,
            client_id=client_id,
            project_id=project_id,
            application_id=application_id,
            session_id=session_id,
        )
        try:
            self._push_tracer(tracer)
            yield tracer
        finally:
            self._pop_tracer(tracer)

    def close(self):
        if self._worker:
            self._worker.kill()
        self.__class__._instance = None

    #
    # Internals
    #

    def _push_tracer(self, tracer: Tracer) -> None:
        tracer._link_parent(self.current_tracer)
        self._tracer.get().append(tracer)

    def _pop_tracer(self, tracer: Tracer) -> None:
        self._tracer.get().remove(tracer)
        tracer._unlink_parent(self.current_tracer)

    def _init_providers(self, providers: List[str]) -> None:
        for provider in providers:
            if provider not in _INSTRUMENTS:
                raise Scope3AIError(
                    f"Could not find tracer for the `{provider}` provider."
                )
            if provider in self._providers and provider not in _RE_INIT_PROVIDERS:
                # already initialized
                continue
            init_func = _INSTRUMENTS[provider]
            init_func()
            self._providers.append(provider)

    def _ensure_worker(self) -> None:
        if not self._worker:
            self._worker = BackgroundWorker(-1)

    def _init_logging(self) -> None:
        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
            handlers=[logging.StreamHandler()],
        )
        logging.getLogger("scope3ai").setLevel(logging.DEBUG)

    def _init_atexit(self):
        @atexit.register
        def _shutdown():
            # do not reinstanciate the singleton here if somehow it was deleted
            scope3ai = Scope3AI._instance
            if not scope3ai:
                return
            if scope3ai._worker and scope3ai._worker._queue:
                logging.debug("Waiting background informations to be processed")
                scope3ai._worker._queue.join()
                logging.debug("Shutting down Scope3AI")

    def _fill_impact_row(
        self,
        row: ImpactRow,
        tracer: Optional[Tracer] = None,
        root_tracer: Optional[Tracer] = None,
    ):
        # fill fields with information we know about
        # One trick is to not set anything on the ImpactRow if it's already set or if the value is None
        # because the model are dumped and exclude the fields unset.
        # If we set a field to None, it will be added for nothing.
        def set_only_if(row, field, *values):
            if getattr(row, field) is not None:
                return
            for value in values:
                if value is not None:
                    setattr(row, field, value)
                    return

        row.request_id = generate_id()
        if root_tracer:
            set_only_if(row, "trace_id", root_tracer.trace_id)
        if row.utc_datetime is None:
            row.utc_datetime = datetime.now(tz=timezone.utc)

        # copy global-only metadata
        set_only_if(
            row,
            "environment",
            self.environment,
        )

        # copy tracer or global metadata

        set_only_if(
            row,
            "managed_service_id",
            row.managed_service_id if row.managed_service_id else "",
        )

        set_only_if(
            row,
            "client_id",
            tracer.client_id if tracer else None,
            self.client_id,
        )
        set_only_if(
            row,
            "project_id",
            tracer.project_id if tracer else None,
            self.project_id,
        )
        set_only_if(
            row,
            "application_id",
            tracer.application_id if tracer else None,
            self.application_id,
        )
        set_only_if(
            row,
            "session_id",
            tracer.session_id if tracer else None,
        )

1	import atexit	4✔
2	import importlib.util	4✔
3	import logging	4✔
4	from contextlib import contextmanager	4✔
5	from contextvars import ContextVar	4✔
6	from datetime import datetime, timezone	4✔
7	from functools import partial	4✔
8	from os import getenv	4✔
9	from typing import List, Optional	4✔
10	from uuid import uuid4	4✔
11
12	from .api.client import AsyncClient, Client	4✔
13	from .api.defaults import DEFAULT_API_URL, DEFAULT_APPLICATION_ID	4✔
14	from .api.tracer import Tracer	4✔
15	from .api.types import ImpactRequest, ImpactResponse, ImpactRow, Scope3AIContext	4✔
16	from .constants import PROVIDERS	4✔
17	from .worker import BackgroundWorker	4✔
18
19	logger = logging.getLogger("scope3ai.lib")	4✔
20
21
22	def init_anthropic_instrumentor() -> None:	4✔
23	if importlib.util.find_spec("anthropic") is not None:	4✔
24	from scope3ai.tracers.anthropic.instrument import AnthropicInstrumentor	4✔
25
26	instrumentor = AnthropicInstrumentor()	4✔
27	instrumentor.instrument()	4✔
28
29
30	def init_cohere_instrumentor() -> None:	4✔
31	if importlib.util.find_spec("cohere") is not None:	4✔
32	from scope3ai.tracers.cohere.instrument import CohereInstrumentor	4✔
33
34	instrumentor = CohereInstrumentor()	4✔
35	instrumentor.instrument()	4✔
36
37
38	def init_openai_instrumentor() -> None:	4✔
39	if importlib.util.find_spec("openai") is not None:	4✔
40	from scope3ai.tracers.openai.instrument import OpenAIInstrumentor	4✔
41
42	instrumentor = OpenAIInstrumentor()	4✔
43	instrumentor.instrument()	4✔
44
45
46	def init_huggingface_hub_instrumentor() -> None:	4✔
47	if importlib.util.find_spec("huggingface_hub") is not None:	4✔
48	from scope3ai.tracers.huggingface.instrument import HuggingfaceInstrumentor	4✔
49
50	instrumentor = HuggingfaceInstrumentor()	4✔
51	instrumentor.instrument()	4✔
52
53
54	def init_google_genai_instrumentor() -> None:	4✔
55	if importlib.util.find_spec("google") is not None:	4✔
56	from scope3ai.tracers.google_genai.instrument import GoogleGenAiInstrumentor	4✔
57
58	instrumentor = GoogleGenAiInstrumentor()	4✔
59	instrumentor.instrument()	4✔
60
61
62	def init_litellm_instrumentor() -> None:	4✔
63	if importlib.util.find_spec("litellm") is not None:	4✔
64	from scope3ai.tracers.litellm.instrument import LiteLLMInstrumentor	4✔
65
66	instrumentor = LiteLLMInstrumentor()	4✔
67	instrumentor.instrument()	4✔
68
69
70	def init_mistral_v1_instrumentor() -> None:	4✔
71	if importlib.util.find_spec("mistralai") is not None:	4✔
72	from scope3ai.tracers.mistralai.instrument import MistralAIInstrumentor	4✔
73
74	instrumentor = MistralAIInstrumentor()	4✔
75	instrumentor.instrument()	4✔
76
77
78	def init_response_instrumentor() -> None:	4✔
79	from scope3ai.response_interceptor.instrument import ResponseInterceptor	4✔
80
81	instrumentor = ResponseInterceptor()	4✔
82	instrumentor.instrument()	4✔
83
84
85	_INSTRUMENTS = {	4✔
86	PROVIDERS.ANTROPIC.value: init_anthropic_instrumentor,
87	PROVIDERS.COHERE.value: init_cohere_instrumentor,
88	PROVIDERS.OPENAI.value: init_openai_instrumentor,
89	PROVIDERS.HUGGINGFACE_HUB.value: init_huggingface_hub_instrumentor,
90	PROVIDERS.GOOGLE_GENAI.value: init_google_genai_instrumentor,
91	PROVIDERS.LITELLM.value: init_litellm_instrumentor,
92	PROVIDERS.MISTRALAI.value: init_mistral_v1_instrumentor,
93	PROVIDERS.RESPONSE.value: init_response_instrumentor,
94	}
95
96	_RE_INIT_PROVIDERS = [PROVIDERS.RESPONSE.value]	4✔
97
98
99	def generate_id() -> str:	4✔
100	return uuid4().hex	4✔
101
102
103	class Scope3AIError(Exception):	4✔
104	pass	4✔
105
106
107	class Scope3AI:	4✔
108	"""
109	Scope3AI tracer class
110
111	This class is a singleton that provides a context manager for tracing
112	inference metadata and submitting impact requests to the Scope3 AI API.
113	"""
114
115	_instance: Optional["Scope3AI"] = None	4✔
116	_tracer: ContextVar[List[Tracer]] = ContextVar("tracer", default=[])	4✔
117	_worker: Optional[BackgroundWorker] = None	4✔
118	_providers: List[str] = []	4✔
119	_keep_tracers: bool = False	4✔
120
121	def __new__(cls, args, *kwargs):	4✔
122	if cls._instance is None:	4✔
123	cls._instance = super(Scope3AI, cls).__new__(cls)	4✔
124	return cls._instance	4✔
125
126	def __init__(self):	4✔
127	self.api_key: Optional[str] = None	4✔
128	self.api_url: Optional[str] = None	4✔
129	self.sync_mode: bool = False	4✔
130	self._sync_client: Optional[Client] = None	4✔
131	self._async_client: Optional[AsyncClient] = None	4✔
132	self.environment: Optional[str] = None	4✔
133	self.client_id: Optional[str] = None	4✔
134	self.project_id: Optional[str] = None	4✔
135	self.application_id: Optional[str] = None	4✔
136
137	@classmethod	4✔
138	def init(	4✔
139	cls,
140	api_key: Optional[str] = None,
141	api_url: Optional[str] = None,
142	sync_mode: bool = False,
143	enable_debug_logging: bool = False,
144	providers: Optional[List[str]] = None,
145	# metadata for scope3
146	environment: Optional[str] = None,
147	client_id: Optional[str] = None,
148	project_id: Optional[str] = None,
149	application_id: Optional[str] = None,
150	) -> "Scope3AI":
151	"""
152	Initialize the Scope3AI SDK with the provided configuration settings.
153
154	Args:
155	api_key (str, optional): The Scope3AI API key. Can be set via `SCOPE3AI_API_KEY`
156	environment variable. Required for authentication.
157	api_url (str, optional): The base URL for the Scope3AI API. Can be set via
158	`SCOPE3AI_API_URL` environment variable. Defaults to standard API URL.
159	sync_mode (bool, optional): If True, the SDK will operate synchronously. Can be
160	set via `SCOPE3AI_SYNC_MODE` environment variable. Defaults to False.
161	enable_debug_logging (bool, optional): Enable debug level logging. Can be set via
162	`SCOPE3AI_DEBUG_LOGGING` environment variable. Defaults to False.
163	providers (List[str], optional): List of providers to instrument. If None,
164	all available providers will be instrumented.
165	environment (str, optional): The environment name (e.g. "production", "staging").
166	Can be set via `SCOPE3AI_ENVIRONMENT` environment variable.
167	client_id (str, optional): Client identifier for grouping traces. Can be set via
168	`SCOPE3AI_CLIENT_ID` environment variable.
169	project_id (str, optional): Project identifier for grouping traces. Can be set via
170	`SCOPE3AI_PROJECT_ID` environment variable.
171	application_id (str, optional): Application identifier. Can be set via
172	`SCOPE3AI_APPLICATION_ID` environment variable. Defaults to "default".
173
174	Returns:
175	Scope3AI: The initialized Scope3AI instance.
176
177	Raises:
178	Scope3AIError: If the instance is already initialized or if required settings are missing.
179	"""
180	if cls._instance is not None:	4✔
181	raise Scope3AIError("Scope3AI is already initialized")	×
182	cls._instance = self = Scope3AI()	4✔
183	self.api_key = api_key or getenv("SCOPE3AI_API_KEY")	4✔
184	self.api_url = api_url or getenv("SCOPE3AI_API_URL") or DEFAULT_API_URL	4✔
185	self.sync_mode = sync_mode or bool(getenv("SCOPE3AI_SYNC_MODE", False))	4✔
186	if not self.api_key:	4✔
187	raise Scope3AIError(	×
188	"The scope3 api_key option must be set either by "
189	"passing the API key to the Scope3AI.init(api_key='xxx') "
190	"or by setting the SCOPE3AI_API_KEY environment variable"
191	)
192	if not self.api_url:	4✔
193	raise Scope3AIError(	×
194	"The api_url option must be set either by "
195	"passing the API URL to the Scope3AI.init(api_url='xxx') "
196	"or by setting the SCOPE3AI_API_URL environment variable"
197	)
198
199	# metadata
200	self.environment = environment or getenv("SCOPE3AI_ENVIRONMENT")	4✔
201	self.client_id = client_id or getenv("SCOPE3AI_CLIENT_ID")	4✔
202	self.project_id = project_id or getenv("SCOPE3AI_PROJECT_ID")	4✔
203	self.application_id = (	4✔
204	application_id
205	or getenv("SCOPE3AI_APPLICATION_ID")
206	or DEFAULT_APPLICATION_ID
207	)
208
209	if enable_debug_logging:	4✔
210	self._init_logging()	4✔
211
212	if providers is None:	4✔
213	providers = list(_INSTRUMENTS.keys())	4✔
214
215	http_client_options = {"api_key": self.api_key, "api_url": self.api_url}	4✔
216	self._sync_client = Client(**http_client_options)	4✔
217	self._async_client = AsyncClient(**http_client_options)	4✔
218	self._init_providers(providers)	4✔
219	self._init_atexit()	4✔
220	return cls._instance	4✔
221
222	@classmethod	4✔
223	def get_instance(cls) -> "Scope3AI":	4✔
224	"""
225	Return the instance of the Scope3AI singleton.
226
227	This method provides access to the default global state of the
228	Scope3AI library. The returned instance can be used to trace
229	inference metadata and submit impact requests to the Scope3 AI
230	API from anywhere in the application.
231
232	Returns:
233	Scope3AI: The singleton instance of the Scope3AI class.
234	"""
235	if not cls._instance:	4✔
236	raise Scope3AIError("Scope3AI is not initialized. Use Scope3AI.init()")	×
237	return cls._instance	4✔
238
239	def submit_impact(	4✔
240	self,
241	impact_row: ImpactRow,
242	) -> Scope3AIContext:
243	"""
244	Submit an impact request to the Scope3 AI API.
245
246	This function sends an impact request represented by the `impact_row`
247	to the Scope3 AI API and optionally returns the response.
248
249	Args:
250	impact_row (ImpactRow): The impact request data
251	that needs to be submitted to the Scope3 AI API.
252
253	Returns:
254	Scope3AIContext: A context object containing the request data and
255	the response from the API.
256	"""
257
258	def submit_impact(	4✔
259	impact_row: ImpactRow,
260	ctx: Scope3AIContext,
261	) -> Optional[ImpactResponse]:
262	assert self._sync_client is not None	4✔
263	response = self._sync_client.get_impact(	4✔
264	content=ImpactRequest(rows=[impact_row]),
265	with_response=True,
266	)
267	ctx.set_impact(response.rows[0])	4✔
268	if ctx._tracer:	4✔
269	ctx._tracer._unlink_trace(ctx)	4✔
270	return response	4✔
271
272	tracer = self.current_tracer	4✔
273	self._fill_impact_row(impact_row, tracer, self.root_tracer)	4✔
274	ctx = Scope3AIContext(request=impact_row)	4✔
275	ctx._tracer = tracer	4✔
276	if tracer:	4✔
277	tracer._link_trace(ctx)	4✔
278
279	if self.sync_mode:	4✔
280	submit_impact(impact_row, ctx=ctx)	4✔
281	return ctx	4✔
282
283	self._ensure_worker()	4✔
284	assert self._worker is not None	4✔
285	self._worker.submit(partial(submit_impact, impact_row=impact_row, ctx=ctx))	4✔
286	return ctx	4✔
287
288	async def asubmit_impact(	4✔
289	self,
290	impact_row: ImpactRow,
291	) -> Scope3AIContext:
292	"""
293	Async version of Scope3AI::submit_impact.
294	"""
295
296	if not self.sync_mode:	4✔
297	# in non sync-mode, it uses the background worker,
298	# and the background worker is not async (does not have to be).
299	# so we just redirect the call to the sync version.
300	return self.submit_impact(impact_row)	4✔
301
302	tracer = self.current_tracer	4✔
303	self._fill_impact_row(impact_row, tracer, self.root_tracer)	4✔
304	ctx = Scope3AIContext(request=impact_row)	4✔
305	ctx._tracer = tracer	4✔
306	if tracer:	4✔
307	tracer._link_trace(ctx)	4✔
308
309	assert self._async_client is not None	4✔
310	response = await self._async_client.get_impact(	4✔
311	content=ImpactRequest(rows=[impact_row]),
312	with_response=True,
313	)
314	ctx.set_impact(response.rows[0])	4✔
315	if tracer:	4✔
316	tracer._unlink_trace(ctx)	4✔
317
318	return ctx	4✔
319
320	@property	4✔
321	def root_tracer(self):	4✔
322	"""
323	Return the root tracer.
324
325	The root tracer is the first tracer in the current execution context
326	(tracer stack). If no tracers are currently active, it returns None.
327
328	Returns:
329	Tracer: The root tracer if available, otherwise None.
330	"""
331	tracers = self._tracer.get()	4✔
332	return tracers[0] if tracers else None	4✔
333
334	@property	4✔
335	def current_tracer(self):	4✔
336	"""
337	Return the current tracer.
338
339	The current tracer is the last tracer in the current execution context
340	(tracer stack). If no tracers are currently active, it returns None.
341
342	Returns:
343	Tracer: The current tracer if available, otherwise None.
344	"""
345	tracers = self._tracer.get()	4✔
346	return tracers[-1] if tracers else None	4✔
347
348	@contextmanager	4✔
349	def trace(	4✔
350	self,
351	keep_traces=False,
352	client_id: Optional[str] = None,
353	project_id: Optional[str] = None,
354	application_id: Optional[str] = None,
355	session_id: Optional[str] = None,
356	):
357	root_tracer = self.root_tracer	4✔
358	if not client_id:	4✔
359	client_id = root_tracer.client_id if root_tracer else self.client_id	4✔
360	if not project_id:	4✔
361	project_id = root_tracer.project_id if root_tracer else self.project_id	4✔
362	if not application_id:	4✔
363	application_id = (	4✔
364	root_tracer.application_id if root_tracer else self.application_id
365	)
366	if not session_id:	4✔
367	session_id = root_tracer.session_id if root_tracer else None	4✔
368	tracer = Tracer(	4✔
369	keep_traces=keep_traces,
370	client_id=client_id,
371	project_id=project_id,
372	application_id=application_id,
373	session_id=session_id,
374	)
375	try:	4✔
376	self._push_tracer(tracer)	4✔
377	yield tracer	4✔
378	finally:
379	self._pop_tracer(tracer)	4✔
380
381	def close(self):	4✔
382	if self._worker:	4✔
383	self._worker.kill()	4✔
384	self.__class__._instance = None	4✔
385
386	#
387	# Internals
388	#
389
390	def _push_tracer(self, tracer: Tracer) -> None:	4✔
391	tracer._link_parent(self.current_tracer)	4✔
392	self._tracer.get().append(tracer)	4✔
393
394	def _pop_tracer(self, tracer: Tracer) -> None:	4✔
395	self._tracer.get().remove(tracer)	4✔
396	tracer._unlink_parent(self.current_tracer)	4✔
397
398	def _init_providers(self, providers: List[str]) -> None:	4✔
399	for provider in providers:	4✔
400	if provider not in _INSTRUMENTS:	4✔
401	raise Scope3AIError(	×
402	f"Could not find tracer for the `{provider}` provider."
403	)
404	if provider in self._providers and provider not in _RE_INIT_PROVIDERS:	4✔
405	# already initialized
406	continue	4✔
407	init_func = _INSTRUMENTS[provider]	4✔
408	init_func()	4✔
409	self._providers.append(provider)	4✔
410
411	def _ensure_worker(self) -> None:	4✔
412	if not self._worker:	4✔
413	self._worker = BackgroundWorker(-1)	4✔
414
415	def _init_logging(self) -> None:	4✔
416	logging.basicConfig(	4✔
417	level=logging.INFO,
418	format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
419	handlers=[logging.StreamHandler()],
420	)
421	logging.getLogger("scope3ai").setLevel(logging.DEBUG)	4✔
422
423	def _init_atexit(self):	4✔
424	@atexit.register	4✔
425	def _shutdown():	4✔
426	# do not reinstanciate the singleton here if somehow it was deleted
427	scope3ai = Scope3AI._instance	×
428	if not scope3ai:	×
429	return	×
430	if scope3ai._worker and scope3ai._worker._queue:	×
431	logging.debug("Waiting background informations to be processed")	×
432	scope3ai._worker._queue.join()	×
433	logging.debug("Shutting down Scope3AI")	×
434
435	def _fill_impact_row(	4✔
436	self,
437	row: ImpactRow,
438	tracer: Optional[Tracer] = None,
439	root_tracer: Optional[Tracer] = None,
440	):
441	# fill fields with information we know about
442	# One trick is to not set anything on the ImpactRow if it's already set or if the value is None
443	# because the model are dumped and exclude the fields unset.
444	# If we set a field to None, it will be added for nothing.
445	def set_only_if(row, field, *values):	4✔
446	if getattr(row, field) is not None:	4✔
447	return	×
448	for value in values:	4✔
449	if value is not None:	4✔
450	setattr(row, field, value)	4✔
451	return	4✔
452
453	row.request_id = generate_id()	4✔
454	if root_tracer:	4✔
455	set_only_if(row, "trace_id", root_tracer.trace_id)	4✔
456	if row.utc_datetime is None:	4✔
457	row.utc_datetime = datetime.now(tz=timezone.utc)	4✔
458
459	# copy global-only metadata
460	set_only_if(	4✔
461	row,
462	"environment",
463	self.environment,
464	)
465
466	# copy tracer or global metadata
467
468	set_only_if(	4✔
469	row,
470	"managed_service_id",
471	row.managed_service_id if row.managed_service_id else "",
472	)
473
474	set_only_if(	4✔
475	row,
476	"client_id",
477	tracer.client_id if tracer else None,
478	self.client_id,
479	)
480	set_only_if(	4✔
481	row,
482	"project_id",
483	tracer.project_id if tracer else None,
484	self.project_id,
485	)
486	set_only_if(	4✔
487	row,
488	"application_id",
489	tracer.application_id if tracer else None,
490	self.application_id,
491	)
492	set_only_if(	4✔
493	row,
494	"session_id",
495	tracer.session_id if tracer else None,
496	)

scope3data / scope3ai-py / 13416015178

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous