14097016956

Committed 27 Mar 2025 01:41AM UTC coverage: 96.23% (+15.7%) from 80.557%

Build # 14097016956

Build Type

Pull #92

github

Specific Base 5758a3

Committed by

dearlordylord

Commit Message

feat(api): client-to-provider dry

Pull Request Pull Request #92: feat: Managed Service Kebabs

Run Details

53 of 55 new or added lines in 11 files covered. (96.36%)

44 existing lines in 10 files now uncovered.

2578 of 2679 relevant lines covered (96.23%)

3.85 hits per line

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

94.74

/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 ImpactRequest, ImpactResponse, ImpactRow, Scope3AIContext
from .constants import CLIENTS
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 = {
    CLIENTS.ANTHROPIC.value: init_anthropic_instrumentor,
    CLIENTS.COHERE.value: init_cohere_instrumentor,
    CLIENTS.OPENAI.value: init_openai_instrumentor,
    CLIENTS.HUGGINGFACE_HUB.value: init_huggingface_hub_instrumentor,
    # TODO current tests use gemini
    CLIENTS.GOOGLE_GENAI.value: init_google_genai_instrumentor,
    CLIENTS.LITELLM.value: init_litellm_instrumentor,
    CLIENTS.MISTRALAI.value: init_mistral_v1_instrumentor,
    CLIENTS.RESPONSE.value: init_response_instrumentor,
}

# TODO what it means / why reinit is allowed here
_RE_INIT_CLIENTS = [CLIENTS.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
    _clients: 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,
        # we have provider_clients and not clients naming here because client also has client_id which is not a [provider] client but a [scope3] client
        provider_clients: 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.
            clients (List[str], optional): List of provider clients to instrument. If None,
                all available provider clients 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()

        clients = provider_clients
        if clients is None:
            clients = 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_clients(clients)
        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_clients(self, clients: List[str]) -> None:
        for client in clients:
            if client not in _INSTRUMENTS:
                raise Scope3AIError(f"Could not find tracer for the `{client}` client.")
            if client in self._clients and client not in _RE_INIT_CLIENTS:
                # already initialized
                continue
            init_func = _INSTRUMENTS[client]
            init_func()
            self._clients.append(client)

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

scope3data / scope3ai-py / 14097016956

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