POC request/response pattern

examples/dispatch.py

@@ -0,0 +1,29 @@
+
+from planet.requests.dispatch import Sync, Async
+from planet.requests import data
+
+get = data.GetSearch("f8737406627546c38f4db4bca5794bc6")
+
+def doit():
+    cl = Sync()
+    print("respx.Response", cl.response(get))
+    print("dict", cl.json(get))
+    search = cl.data(get)
+    print("search class", search)
+    print(search.name, search.search_type)
+
+
+async def doit_async():
+    cl = Async()
+    print("respx.Response", await cl.response(get))
+    print("dict", await cl.json(get))
+    search = await cl.data(get)
+    print("search class", search)
+    print(search.name, search.search_type)
+
+if __name__ == "__main__":
+    import asyncio, sys
+    if sys.argv[1] == "sync":
+        doit()
+    else:
+        asyncio.run(doit_async())

planet/requests/data/__init__.py

@@ -0,0 +1,33 @@
+import httpx
+from ..request import Simple
+from dataclasses import dataclass
+
+@dataclass
+class Search:
+    """
+    Documentation specific to the response (partial for POC)
+    """
+
+    name: str
+    search_type: str
+
+class GetSearch(Simple[Search]):
+    """
+    Documentation specific to GetSearch operation
+    """
+
+    _response_class = Search
+    _url_format = 'https://api.planet.com/data/v1/searches/{id}'
+
+    def __init__(self, id):
+        """
+        Documentation specific to operation parameters
+        """
+        self._url_args = {"id": id}
+
+    def _response(self, response: httpx.Response):
+        json = response.json()
+        # because the model is not complete, need to manually call keywords
+        # ideally the Search dataclass would have all the fields declared
+        # and this override wouldn't be needed (see Simple._response)
+        return Search(name=json["name"], search_type=json["search_type"])

planet/requests/dispatch.py

@@ -0,0 +1,49 @@
+from typing import Optional, cast, Any
+import httpx
+from abc import ABC
+
+from planet.http import Session
+from .request import Request
+
+
+class _Base(ABC):
+    def __init__(self, session: Optional[Session] = None):
+        self._session = session or Session()
+        acl = self._session._client
+        # @todo use real client/session
+        self.client = httpx.Client(
+            auth=acl.auth,
+            headers=acl.headers,
+            base_url="https://api.planet.com",
+            transport=httpx.HTTPTransport(retries=3)
+        )
+
+
+class Sync(_Base):
+
+    def response(self, r: Request) -> httpx.Response:
+        req = r._request()
+        resp = self.client.send(req)
+        return resp.raise_for_status()
+
+    def json(self, r: Request) -> Any:
+        return self.response(r).json()
+
+    def data[T](self, r: Request[T]) -> T:
+        return cast(T, r._response(self.response(r)))
+
+
+class Async(_Base):
+
+    async def response(self, r: Request) -> httpx.Response:
+        req = r._request()
+        resp = self.client.send(req)
+        return resp.raise_for_status()
+
+    async def json(self, r: Request) -> Any:
+        resp = await self.response(r)
+        return resp.json()
+
+    async def data[T](self, r: Request[T]) -> T:
+        resp = await self.response(r)
+        return cast(T, r._response(resp))

planet/requests/request.py

@@ -0,0 +1,43 @@
+from typing import TypeVar, ClassVar, Any
+import httpx
+from abc import ABC, abstractmethod
+
+
+ResponseType = TypeVar("ResponseType", bound=type)
+
+class Request[ResponseType](ABC):
+
+    _method: str = "GET"
+    _url_format: str
+    _response_class: ClassVar[type[ResponseType]]
+
+    @abstractmethod
+    def _request(self) -> httpx.Request:
+        pass
+
+    @abstractmethod
+    def _url(self) -> str:
+        pass
+
+    @abstractmethod
+    def _response(self, response: httpx.Response) -> ResponseType:
+        pass
+
+
+class Simple[ResponseType](Request[ResponseType]):
+
+    _url_args: dict[str, Any]
+    _params: dict[str, Any]
+
+    def _url(self):
+        return self._url_format.format(**self._url_args)
+
+    def _request(self) -> httpx.Request:
+        return httpx.Request(
+            method=self._method,
+            url=self._url(),
+            params=getattr(self, "_params", None),
+        )
+
+    def _response(self, response: httpx.Response) -> ResponseType:
+        return self._response_class(**response.json())