12877283541

Committed 20 Jan 2025 11:39PM UTC coverage: 95.796% (+15.2%) from 80.557%

Build # 12877283541

Build Type

Pull #70

github

Specific Base bc1070

Committed by

kevdevg

Commit Message

fix: variable name

Pull Request Pull Request #70: feat(litellm): multimodal

Run Details

101 of 105 new or added lines in 14 files covered. (96.19%)

45 existing lines in 9 files now uncovered.

2142 of 2236 relevant lines covered (95.8%)

3.83 hits per line

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

81.71

/scope3ai/api/client.py

from os import getenv
from typing import List, Optional, Union

import httpx
from pydantic import BaseModel

from .defaults import DEFAULT_API_URL
from .types import (
    Family,
    GPUResponse,
    ImpactRequest,
    ImpactResponse,
    ImpactRow,
    ModelResponse,
    NodeResponse,
)


class Scope3AIError(Exception):
    pass


class ClientBase:
    def __init__(self, api_key: str = None, api_url: str = None) -> None:
        self.api_key = api_key or getenv("SCOPE3AI_API_KEY")
        self.api_url = api_url or getenv("SCOPE3AI_API_URL") or DEFAULT_API_URL
        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"
            )

    @property
    def client(self) -> Union[httpx.Client, httpx.AsyncClient]:
        """
        Obtain an httpx client for synchronous or asynchronous operation
        with the necessary authentication headers included.
        """
        if not hasattr(self, "_client"):
            self._client = self.create_client()
        return self._client


class ClientCommands:
    def model(
        self,
        family: Optional[Family] = None,
        with_response: Optional[bool] = True,
    ) -> ModelResponse:
        """
        List models
        """
        params = {}
        if family:
            params["family"] = family
        return self.execute_request(
            "/model",
            method="GET",
            params=params,
            response_model=ModelResponse,
            with_response=with_response,
        )

    def gpu(
        self,
        with_response: Optional[bool] = True,
    ) -> GPUResponse:
        """
        List GPUs
        """
        return self.execute_request(
            "/gpu",
            method="GET",
            response_model=GPUResponse,
            with_response=with_response,
        )

    def node(
        self,
        service: Optional[str] = None,
        cloud: Optional[str] = None,
        custom: Optional[bool] = None,
        gpu: Optional[str] = None,
        instance: Optional[str] = None,
        with_response: Optional[bool] = True,
    ) -> NodeResponse:
        """
        List nodes
        """
        params = {}
        if service is not None:
            params["service"] = service
        if cloud is not None:
            params["cloud"] = cloud
        if custom is not None:
            params["custom"] = custom
        if gpu is not None:
            params["gpu"] = gpu
        if instance is not None:
            params["instance"] = instance
        return self.execute_request(
            "/node",
            method="GET",
            params=params,
            response_model=NodeResponse,
            with_response=with_response,
        )

    def impact(
        self,
        rows: List[ImpactRow],
        debug: Optional[bool] = None,
        with_response: Optional[bool] = True,
    ) -> ImpactResponse:
        """
        Get impact metrics for a task
        """
        params = {}
        if debug is not None:
            params["debug"] = debug
        json_body = ImpactRequest(rows=rows).model_dump(
            mode="json",
            exclude_unset=True,
        )
        return self.execute_request(
            "/v1/impact",
            method="POST",
            params=params,
            json=json_body,
            response_model=ImpactResponse,
            with_response=with_response,
        )


class Client(ClientBase, ClientCommands):
    """
    Synchronous Client to the Scope3AI HTTP API
    """

    def create_client(self):
        return httpx.Client(headers={"Authorization": f"Bearer {self.api_key}"})

    def execute_request(
        self,
        url: str,
        method="GET",
        params: Optional[dict] = None,
        json: Optional[dict] = None,
        response_model: Optional[BaseModel] = None,
        with_response: Optional[bool] = True,
    ):
        full_url = self.api_url + url
        kwargs = {}
        if params:
            kwargs["params"] = params
        if json:
            kwargs["json"] = json
        response = self.client.request(method, full_url, **kwargs)
        response.raise_for_status()
        if not with_response:
            return
        if response_model:
            return response_model.model_validate(response.json())
        return response.json()


class AsyncClient(ClientBase, ClientCommands):
    """
    Asynchronous Client to the Scope3AI HTTP API
    """

    def create_client(self):
        return httpx.AsyncClient(headers={"Authorization": f"Bearer {self.api_key}"})

    async def execute_request(
        self,
        url: str,
        method="GET",
        params: Optional[dict] = None,
        json: Optional[dict] = None,
        response_model: Optional[BaseModel] = None,
        with_response: Optional[bool] = True,
    ):
        full_url = self.api_url + url
        kwargs = {}
        if params:
            kwargs["params"] = params
        if json:
            kwargs["json"] = json
        response = await self.client.request(method, full_url, **kwargs)
        response.raise_for_status()
        if not with_response:
            return
        if response_model:
            return response_model.model_validate(response.json())
        return response.json()

1	from os import getenv	4✔
2	from typing import List, Optional, Union	4✔
3
4	import httpx	4✔
5	from pydantic import BaseModel	4✔
6
7	from .defaults import DEFAULT_API_URL	4✔
8	from .types import (	4✔
9	Family,
10	GPUResponse,
11	ImpactRequest,
12	ImpactResponse,
13	ImpactRow,
14	ModelResponse,
15	NodeResponse,
16	)
17
18
19	class Scope3AIError(Exception):	4✔
20	pass	4✔
21
22
23	class ClientBase:	4✔
24	def __init__(self, api_key: str = None, api_url: str = None) -> None:	4✔
25	self.api_key = api_key or getenv("SCOPE3AI_API_KEY")	4✔
26	self.api_url = api_url or getenv("SCOPE3AI_API_URL") or DEFAULT_API_URL	4✔
27	if not self.api_key:	4✔
UNCOV 28	raise Scope3AIError(	×
29	"The scope3 api_key option must be set either by "
30	"passing the API key to the Scope3AI.init(api_key='xxx') "
31	"or by setting the SCOPE3AI_API_KEY environment variable"
32	)
33	if not self.api_url:	4✔
UNCOV 34	raise Scope3AIError(	×
35	"The api_url option must be set either by "
36	"passing the API URL to the Scope3AI.init(api_url='xxx') "
37	"or by setting the SCOPE3AI_API_URL environment variable"
38	)
39
40	@property	4✔
41	def client(self) -> Union[httpx.Client, httpx.AsyncClient]:	4✔
42	"""
43	Obtain an httpx client for synchronous or asynchronous operation
44	with the necessary authentication headers included.
45	"""
46	if not hasattr(self, "_client"):	4✔
47	self._client = self.create_client()	4✔
48	return self._client	4✔
49
50
51	class ClientCommands:	4✔
52	def model(	4✔
53	self,
54	family: Optional[Family] = None,
55	with_response: Optional[bool] = True,
56	) -> ModelResponse:
57	"""
58	List models
59	"""
60	params = {}	4✔
61	if family:	4✔
62	params["family"] = family	×
63	return self.execute_request(	4✔
64	"/model",
65	method="GET",
66	params=params,
67	response_model=ModelResponse,
68	with_response=with_response,
69	)
70
71	def gpu(	4✔
72	self,
73	with_response: Optional[bool] = True,
74	) -> GPUResponse:
75	"""
76	List GPUs
77	"""
78	return self.execute_request(	4✔
79	"/gpu",
80	method="GET",
81	response_model=GPUResponse,
82	with_response=with_response,
83	)
84
85	def node(	4✔
86	self,
87	service: Optional[str] = None,
88	cloud: Optional[str] = None,
89	custom: Optional[bool] = None,
90	gpu: Optional[str] = None,
91	instance: Optional[str] = None,
92	with_response: Optional[bool] = True,
93	) -> NodeResponse:
94	"""
95	List nodes
96	"""
97	params = {}	4✔
98	if service is not None:	4✔
99	params["service"] = service	×
100	if cloud is not None:	4✔
101	params["cloud"] = cloud	×
102	if custom is not None:	4✔
UNCOV 103	params["custom"] = custom	×
104	if gpu is not None:	4✔
UNCOV 105	params["gpu"] = gpu	×
106	if instance is not None:	4✔
UNCOV 107	params["instance"] = instance	×
108	return self.execute_request(	4✔
109	"/node",
110	method="GET",
111	params=params,
112	response_model=NodeResponse,
113	with_response=with_response,
114	)
115
116	def impact(	4✔
117	self,
118	rows: List[ImpactRow],
119	debug: Optional[bool] = None,
120	with_response: Optional[bool] = True,
121	) -> ImpactResponse:
122	"""
123	Get impact metrics for a task
124	"""
125	params = {}	4✔
126	if debug is not None:	4✔
UNCOV 127	params["debug"] = debug	×
128	json_body = ImpactRequest(rows=rows).model_dump(	4✔
129	mode="json",
130	exclude_unset=True,
131	)
132	return self.execute_request(	4✔
133	"/v1/impact",
134	method="POST",
135	params=params,
136	json=json_body,
137	response_model=ImpactResponse,
138	with_response=with_response,
139	)
140
141
142	class Client(ClientBase, ClientCommands):	4✔
143	"""
144	Synchronous Client to the Scope3AI HTTP API
145	"""
146
147	def create_client(self):	4✔
148	return httpx.Client(headers={"Authorization": f"Bearer {self.api_key}"})	4✔
149
150	def execute_request(	4✔
151	self,
152	url: str,
153	method="GET",
154	params: Optional[dict] = None,
155	json: Optional[dict] = None,
156	response_model: Optional[BaseModel] = None,
157	with_response: Optional[bool] = True,
158	):
159	full_url = self.api_url + url	4✔
160	kwargs = {}	4✔
161	if params:	4✔
UNCOV 162	kwargs["params"] = params	×
163	if json:	4✔
164	kwargs["json"] = json	4✔
165	response = self.client.request(method, full_url, **kwargs)	4✔
166	response.raise_for_status()	4✔
167	if not with_response:	4✔
UNCOV 168	return	×
169	if response_model:	4✔
170	return response_model.model_validate(response.json())	4✔
UNCOV 171	return response.json()	×
172
173
174	class AsyncClient(ClientBase, ClientCommands):	4✔
175	"""
176	Asynchronous Client to the Scope3AI HTTP API
177	"""
178
179	def create_client(self):	4✔
180	return httpx.AsyncClient(headers={"Authorization": f"Bearer {self.api_key}"})	4✔
181
182	async def execute_request(	4✔
183	self,
184	url: str,
185	method="GET",
186	params: Optional[dict] = None,
187	json: Optional[dict] = None,
188	response_model: Optional[BaseModel] = None,
189	with_response: Optional[bool] = True,
190	):
191	full_url = self.api_url + url	4✔
192	kwargs = {}	4✔
193	if params:	4✔
194	kwargs["params"] = params	×
195	if json:	4✔
196	kwargs["json"] = json	4✔
197	response = await self.client.request(method, full_url, **kwargs)	4✔
198	response.raise_for_status()	4✔
199	if not with_response:	4✔
200	return	×
201	if response_model:	4✔
202	return response_model.model_validate(response.json())	4✔
UNCOV 203	return response.json()	×

scope3data / scope3ai-py / 12877283541

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