added data client

Author: Kraysent

hyperleda/common.py

@@ -0,0 +1,21 @@
+from typing import Any
+
+import requests
+
+from hyperleda import error, utils
+
+
+def request(method: str, url: str, query: dict[str, Any] | None = None, stream: bool = False) -> requests.Response:
+    kwargs = {}
+    if query is not None:
+        kwargs["params"] = utils.clean_dict(query)
+    if stream:
+        kwargs["stream"] = True
+    response = requests.request(method, url, **kwargs)
+    if not response.ok:
+        try:
+            msg = error.APIError.from_dict(response.json())
+        except Exception:
+            msg = error.APIError(response.status_code, response.text)
+        raise msg
+    return response

hyperleda/data/client.py

@@ -0,0 +1,53 @@
+import dataclasses
+from typing import Any
+
+from hyperleda import common, config
+from hyperleda.data import model
+
+
+class HyperLedaDataClient:
+    """
+    Client for the HyperLeda data API, providing access to public query endpoints.
+    """
+
+    def __init__(self, endpoint: str = config.DEFAULT_ENDPOINT):
+        self.endpoint = endpoint
+
+    def _request(self, method: str, path: str, query: dict[str, Any] | None = None, stream: bool = False) -> Any:
+        return common.request(method, f"{self.endpoint}{path}", query=query, stream=stream)
+
+    def query_simple(self, req: model.QuerySimpleRequestSchema) -> model.QuerySimpleResponseSchema:
+        """
+        Query data about objects using simple parameters (AND logic).
+        """
+        response = self._request(
+            "GET",
+            "/api/v1/query/simple",
+            query=dataclasses.asdict(req),
+        )
+        data = response.json()
+        return model.QuerySimpleResponseSchema(**data["data"])
+
+    def query(self, req: model.QueryRequestSchema) -> model.QueryResponseSchema:
+        """
+        Query data about objects using a query string (functions and operators).
+        """
+        response = self._request(
+            "GET",
+            "/api/v1/query",
+            query=dataclasses.asdict(req),
+        )
+        data = response.json()
+        return model.QueryResponseSchema(**data["data"])
+
+    def query_fits(self, req: model.FITSRequestSchema) -> bytes:
+        """
+        Query data about objects and return as FITS file (binary).
+        """
+        response = self._request(
+            "GET",
+            "/api/v1/query/fits",
+            query=dataclasses.asdict(req),
+            stream=True,
+        )
+        return response.content

hyperleda/data/model.py

@@ -0,0 +1,80 @@
+# generated by datamodel-codegen:
+#   filename:  swagger.json
+#   timestamp: 2025-06-07T08:18:05+00:00
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+
+class PageSize(Enum):
+    integer_10 = 10
+    integer_25 = 25
+    integer_50 = 50
+    integer_100 = 100
+
+
+@dataclass
+class QuerySimpleRequestSchema:
+    pgcs: list[int] | None = None
+    ra: float | None = None
+    dec: float | None = None
+    radius: float | None = None
+    name: str | None = None
+    cz: float | None = None
+    cz_err_percent: float | None = None
+    page_size: PageSize | None = PageSize.integer_25
+    page: int | None = 0
+
+
+@dataclass
+class PGCObject:
+    pgc: int | None = None
+    catalogs: dict[str, dict[str, Any]] | None = None
+
+
+@dataclass
+class QuerySimpleResponseSchema:
+    objects: list[PGCObject] | None = None
+
+
+@dataclass
+class QueryRequestSchema:
+    q: str | None = None
+    page_size: PageSize | None = PageSize.integer_25
+    page: int | None = 0
+
+
+@dataclass
+class QueryResponseSchema:
+    objects: list[PGCObject] | None = None
+
+
+class PageSize2(Enum):
+    integer_10 = 10
+    integer_25 = 25
+    integer_50 = 50
+    integer_100 = 100
+    integer_1000 = 1000
+    integer_10000 = 10000
+    integer_100000 = 100000
+
+
+@dataclass
+class FITSRequestSchema:
+    pgcs: list[int] | None = None
+    ra: float | None = None
+    dec: float | None = None
+    radius: float | None = None
+    name: str | None = None
+    cz: float | None = None
+    cz_err_percent: float | None = None
+    page_size: PageSize2 | None = PageSize2.integer_25
+    page: int | None = 0
+
+
+@dataclass
+class Schema:
+    pass

makefile

@@ -11,12 +11,18 @@ build:
 	$(PYTHON) -m build .
 
 # to use the model one needs to start the server on localhost:8080
-generate-model:
+generate-admin-model:
 	mkdir -p hyperleda/gen
 	curl http://localhost:8080/admin/api/docs/swagger.json > hyperleda/gen/swagger.json
 	datamodel-codegen --input hyperleda/gen/swagger.json --output hyperleda/model.py --output-model-type dataclasses.dataclass --input-file-type openapi
 	make fix
 
+generate-data-model:
+	mkdir -p hyperleda/data/gen
+	curl http://localhost:8080/api/docs/swagger.json > hyperleda/data/gen/swagger.json
+	datamodel-codegen --input hyperleda/data/gen/swagger.json --output hyperleda/data/model.py --output-model-type dataclasses.dataclass --input-file-type openapi
+	make fix
+
 # Testing
 
 check: check-format check-lint