12920205472

Committed 23 Jan 2025 01:12AM UTC coverage: 95.993% (+15.4%) from 80.557%

Build # 12920205472

Build Type

Pull #74

github

Specific Base 9af6ef

Committed by

tito

Commit Message

fix: try fixing test again

Pull Request Pull Request #74: feat(metadata): include many metadata accessible at global or tracer level

Run Details

49 of 50 new or added lines in 4 files covered. (98.0%)

53 existing lines in 10 files now uncovered.

2204 of 2296 relevant lines covered (95.99%)

3.84 hits per line

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

93.18

/scope3ai/lib.py

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)
            self._fill_impact_row_for_tracer(impact_row, tracer, self.root_tracer)

        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✔
UNCOV 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✔
UNCOV 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✔
UNCOV 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✔
UNCOV 267	tracer._link_trace(ctx)	×
NEW 268	self._fill_impact_row_for_tracer(impact_row, tracer, self.root_tracer)	×
269
270	response = await self._async_client.impact(	4✔
271	rows=[impact_row],
272	with_response=True,
273	)
274	ctx.set_impact(response.rows[0])	4✔
275	if tracer:	4✔
UNCOV 276	tracer.add_impact(response.rows[0])	×
UNCOV 277	tracer._unlink_trace(ctx)	×
278
279	return ctx	4✔
280
281	@property	4✔
282	def root_tracer(self):	4✔
283	"""
284	Return the root tracer.
285
286	The root tracer is the first tracer in the current execution context
287	(tracer stack). If no tracers are currently active, it returns None.
288
289	Returns:
290	Tracer: The root tracer if available, otherwise None.
291	"""
292	tracers = self._tracer.get()	4✔
293	return tracers[0] if tracers else None	4✔
294
295	@property	4✔
296	def current_tracer(self):	4✔
297	"""
298	Return the current tracer.
299
300	The current tracer is the last tracer in the current execution context
301	(tracer stack). If no tracers are currently active, it returns None.
302
303	Returns:
304	Tracer: The current tracer if available, otherwise None.
305	"""
306	tracers = self._tracer.get()	4✔
307	return tracers[-1] if tracers else None	4✔
308
309	@contextmanager	4✔
310	def trace(	4✔
311	self,
312	keep_traces=False,
313	client_id: Optional[str] = None,
314	project_id: Optional[str] = None,
315	application_id: Optional[str] = None,
316	session_id: Optional[str] = None,
317	):
318	root_tracer = self.root_tracer	4✔
319	if not client_id:	4✔
320	client_id = root_tracer.client_id if root_tracer else self.client_id	4✔
321	if not project_id:	4✔
322	project_id = root_tracer.project_id if root_tracer else self.project_id	4✔
323	if not application_id:	4✔
324	application_id = (	4✔
325	root_tracer.application_id if root_tracer else self.application_id
326	)
327	if not session_id:	4✔
328	session_id = root_tracer.session_id if root_tracer else None	4✔
329	tracer = Tracer(	4✔
330	keep_traces=keep_traces,
331	client_id=client_id,
332	project_id=project_id,
333	application_id=application_id,
334	session_id=session_id,
335	)
336	try:	4✔
337	self._push_tracer(tracer)	4✔
338	yield tracer	4✔
339	finally:
340	self._pop_tracer(tracer)	4✔
341
342	def close(self):	4✔
343	if self._worker:	4✔
344	self._worker.kill()	4✔
345	self.__class__._instance = None	4✔
346
347	#
348	# Internals
349	#
350
351	def _push_tracer(self, tracer: Tracer) -> None:	4✔
352	tracer._link_parent(self.current_tracer)	4✔
353	self._tracer.get().append(tracer)	4✔
354
355	def _pop_tracer(self, tracer: Tracer) -> None:	4✔
356	self._tracer.get().remove(tracer)	4✔
357	tracer._unlink_parent(self.current_tracer)	4✔
358
359	def _init_providers(self, providers: List[str]) -> None:	4✔
360	"""Initialize the specified providers."""
361	for provider in providers:	4✔
362	if provider not in _INSTRUMENTS:	4✔
UNCOV 363	raise Scope3AIError(	×
364	f"Could not find tracer for the `{provider}` provider."
365	)
366	if provider in self._providers and provider not in _RE_INIT_PROVIDERS:	4✔
367	# already initialized
368	continue	4✔
369	init_func = _INSTRUMENTS[provider]	4✔
370	init_func()	4✔
371	self._providers.append(provider)	4✔
372
373	def _ensure_worker(self) -> None:	4✔
374	if not self._worker:	4✔
375	self._worker = BackgroundWorker(-1)	4✔
376
377	def _init_logging(self) -> None:	4✔
378	logging.basicConfig(	4✔
379	level=logging.INFO,
380	format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
381	handlers=[logging.StreamHandler()],
382	)
383	logging.getLogger("scope3ai").setLevel(logging.DEBUG)	4✔
384
385	def _init_atexit(self):	4✔
386	@atexit.register	4✔
387	def _shutdown():	4✔
388	# do not reinstanciate the singleton here if somehow it was deleted
UNCOV 389	scope3ai = Scope3AI._instance	×
UNCOV 390	if not scope3ai:	×
UNCOV 391	return	×
UNCOV 392	if scope3ai._worker and scope3ai._worker._queue:	×
UNCOV 393	logging.debug("Waiting background informations to be processed")	×
UNCOV 394	scope3ai._worker._queue.join()	×
UNCOV 395	logging.debug("Shutting down Scope3AI")	×
396
397	def _fill_impact_row(	4✔
398	self,
399	row: ImpactRow,
400	tracer: Optional[Tracer] = None,
401	root_tracer: Optional[Tracer] = None,
402	):
403	# fill fields with information we know about
404	# One trick is to not set anything on the ImpactRow if it's already set or if the value is None
405	# because the model are dumped and exclude the fields unset.
406	# If we set a field to None, it will be added for nothing.
407	def set_only_if(row, field, *values):	4✔
408	if getattr(row, field) is not None:	4✔
409	return	4✔
410	for value in values:	4✔
411	if value is not None:	4✔
412	setattr(row, field, value)	4✔
413	return	4✔
414
415	row.request_id = generate_id()	4✔
416	if root_tracer:	4✔
417	set_only_if(row, "trace_id", root_tracer.trace_id)	4✔
418	if row.utc_datetime is None:	4✔
419	row.utc_datetime = datetime.now(tz=timezone.utc)	4✔
420
421	# copy global-only metadata
422	set_only_if(	4✔
423	row,
424	"environment",
425	self.environment,
426	)
427
428	# copy tracer or global metadata
429	set_only_if(	4✔
430	row,
431	"client_id",
432	tracer.client_id if tracer else None,
433	self.client_id,
434	)
435	set_only_if(	4✔
436	row,
437	"project_id",
438	tracer.project_id if tracer else None,
439	self.project_id,
440	)
441	set_only_if(	4✔
442	row,
443	"application_id",
444	tracer.application_id if tracer else None,
445	self.application_id,
446	)
447	set_only_if(	4✔
448	row,
449	"session_id",
450	tracer.session_id if tracer else None,
451	)

scope3data / scope3ai-py / 12920205472

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