12938282577

import atexit
import importlib.metadata
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 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_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.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":
        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.
        """
        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]:
            response = self._sync_client.impact(
                rows=[impact_row],
                with_response=True,
            )
            ctx.set_impact(response.rows[0])
            if ctx._tracer:
                ctx._tracer.add_impact(response.rows[0])
                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()
        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)

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

scope3data / scope3ai-py / 12938282577

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