13040707031

Committed 29 Jan 2025 09:23PM UTC coverage: 96.412% (+15.9%) from 80.557%

Build # 13040707031

Build Type

Pull #84

github

Specific Base 24322d

Committed by

kevdevg

Commit Message

feat: multimodal output for openain/litellm

Pull Request Pull Request #84: feat: multimodal output audio for OpenAi and Litellm

Run Details

33 of 34 new or added lines in 3 files covered. (97.06%)

54 existing lines in 10 files now uncovered.

2472 of 2564 relevant lines covered (96.41%)

3.85 hits per line

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

94.62

/scope3ai/lib.py

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

scope3data / scope3ai-py / 13040707031

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