12953930249

Committed 24 Jan 2025 04:50PM UTC coverage: 96.405% (+15.8%) from 80.557%

Build # 12953930249

Build Type

Pull #78

github

Specific Base 89ada1

Committed by

web-flow

Commit Message

Merge 1e5564797 into 0cfbba85d

Pull Request Pull Request #78: feat(api): synchronize api, fixes pyright issues and api example

Run Details

69 of 71 new or added lines in 2 files covered. (97.18%)

48 existing lines in 10 files now uncovered.

2440 of 2531 relevant lines covered (96.4%)

3.85 hits per line

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

80.46

/scope3ai/api/client.py

from os import getenv
from typing import Any, List, Optional, TypeVar

import httpx
from pydantic import BaseModel

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

ClientType = TypeVar("ClientType", httpx.Client, httpx.AsyncClient)


class Scope3AIError(Exception):
    pass


class ClientBase:
    def __init__(
        self,
        api_key: Optional[str] = None,
        api_url: Optional[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) -> ClientType:
        """
        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

    def create_client(self) -> ClientType:
        raise NotImplementedError


class ClientCommands:
    def execute_request(self, *args, **kwargs) -> Any:
        raise NotImplementedError

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

scope3data / scope3ai-py / 12953930249

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