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

85.48

/scope3ai/api/client.py

from os import getenv
from typing import Optional, TypeVar

import httpx
from pydantic import BaseModel

from .commandsgen import ClientCommands
from .defaults import DEFAULT_API_URL

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


class Scope3AIError(Exception):
    pass


class ClientBase:
    """
    Base client class for communicating with the Scope3AI HTTP API.

    Handles authentication and provides core request functionality for both
    synchronous and asynchronous clients.

    Attributes:
        api_key (Optional[str]): API key for authentication, can be passed in or read from env var
        api_url (Optional[str]): URL for the API, defaults to production endpoint
    """

    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 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:
            if isinstance(json, BaseModel):
                json = json.model_dump(mode="json", exclude_unset=True)
            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:
            if isinstance(json, BaseModel):
                json = json.model_dump(mode="json", exclude_unset=True)
            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 Optional, TypeVar	4✔
3
4	import httpx	4✔
5	from pydantic import BaseModel	4✔
6
7	from .commandsgen import ClientCommands	4✔
8	from .defaults import DEFAULT_API_URL	4✔
9
10	ClientType = TypeVar("ClientType", httpx.Client, httpx.AsyncClient)	4✔
11
12
13	class Scope3AIError(Exception):	4✔
14	pass	4✔
15
16
17	class ClientBase:	4✔
18	"""
19	Base client class for communicating with the Scope3AI HTTP API.
20
21	Handles authentication and provides core request functionality for both
22	synchronous and asynchronous clients.
23
24	Attributes:
25	api_key (Optional[str]): API key for authentication, can be passed in or read from env var
26	api_url (Optional[str]): URL for the API, defaults to production endpoint
27	"""
28
29	def __init__(	4✔
30	self,
31	api_key: Optional[str] = None,
32	api_url: Optional[str] = None,
33	) -> None:
34	self.api_key = api_key or getenv("SCOPE3AI_API_KEY")	4✔
35	self.api_url = api_url or getenv("SCOPE3AI_API_URL") or DEFAULT_API_URL	4✔
36	if not self.api_key:	4✔
UNCOV 37	raise Scope3AIError(	×
38	"The scope3 api_key option must be set either by "
39	"passing the API key to the Scope3AI.init(api_key='xxx') "
40	"or by setting the SCOPE3AI_API_KEY environment variable"
41	)
42	if not self.api_url:	4✔
UNCOV 43	raise Scope3AIError(	×
44	"The api_url option must be set either by "
45	"passing the API URL to the Scope3AI.init(api_url='xxx') "
46	"or by setting the SCOPE3AI_API_URL environment variable"
47	)
48
49	@property	4✔
50	def client(self) -> ClientType:	4✔
51	"""
52	Obtain an httpx client for synchronous or asynchronous operation
53	with the necessary authentication headers included.
54	"""
55	if not hasattr(self, "_client"):	4✔
56	self._client = self.create_client()	4✔
57	return self._client	4✔
58
59	def create_client(self) -> ClientType:	4✔
UNCOV 60	raise NotImplementedError	×
61
62
63	class Client(ClientBase, ClientCommands):	4✔
64	"""
65	Synchronous Client to the Scope3AI HTTP API
66	"""
67
68	def create_client(self) -> httpx.Client:	4✔
69	return httpx.Client(headers={"Authorization": f"Bearer {self.api_key}"})	4✔
70
71	def execute_request(	4✔
72	self,
73	url: str,
74	method="GET",
75	params: Optional[dict] = None,
76	json: Optional[dict] = None,
77	response_model: Optional[BaseModel] = None,
78	with_response: Optional[bool] = True,
79	):
80	full_url = self.api_url + url	4✔
81	kwargs = {}	4✔
82	if params:	4✔
UNCOV 83	kwargs["params"] = params	×
84	if json:	4✔
85	if isinstance(json, BaseModel):	4✔
86	json = json.model_dump(mode="json", exclude_unset=True)	4✔
87	kwargs["json"] = json	4✔
88	response = self.client.request(method, full_url, **kwargs)	4✔
89	response.raise_for_status()	4✔
90	if not with_response:	4✔
UNCOV 91	return	×
92	if response_model:	4✔
93	return response_model.model_validate(response.json())	4✔
UNCOV 94	return response.json()	×
95
96
97	class AsyncClient(ClientBase, ClientCommands):	4✔
98	"""
99	Asynchronous Client to the Scope3AI HTTP API
100	"""
101
102	def create_client(self):	4✔
103	return httpx.AsyncClient(headers={"Authorization": f"Bearer {self.api_key}"})	4✔
104
105	async def execute_request(	4✔
106	self,
107	url: str,
108	method="GET",
109	params: Optional[dict] = None,
110	json: Optional[dict] = None,
111	response_model: Optional[BaseModel] = None,
112	with_response: Optional[bool] = True,
113	):
114	full_url = self.api_url + url	4✔
115	kwargs = {}	4✔
116	if params:	4✔
UNCOV 117	kwargs["params"] = params	×
118	if json:	4✔
119	if isinstance(json, BaseModel):	4✔
120	json = json.model_dump(mode="json", exclude_unset=True)	4✔
121	kwargs["json"] = json	4✔
122	response = await self.client.request(method, full_url, **kwargs)	4✔
123	response.raise_for_status()	4✔
124	if not with_response:	4✔
UNCOV 125	return	×
126	if response_model:	4✔
127	return response_model.model_validate(response.json())	4✔
UNCOV 128	return response.json()	×

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