Design Generic function for the getJP2Image endpoint

.gitignore

@@ -89,6 +89,7 @@ ipython_config.py
 #   However, in case of collaboration, if having platform-specific dependencies or dependencies
 #   having no cross-platform support, pipenv may install dependencies that don't work, or not
 #   install all needed dependencies.
+Pipfile
 #Pipfile.lock
 
 # PEP 582; used by e.g. github.com/David-OConnor/pyflow

hvpy/__init__.py

@@ -1,2 +1 @@
-from .api import *  # NOQA
 from .version import __version__

hvpy/api_groups/jpeg2000/get_jp2_image.py

@@ -0,0 +1,47 @@
+from datetime import datetime
+
+from pydantic import Field, validator
+
+from hvpy.io import HvpyParameters, OutputType
+
+
+class getJP2ImageInputParameters(HvpyParameters):
+    """
+    Handles the input parameters of the `getJP2Image API.
+
+    <https://api.helioviewer.org/docs/v2/api/api_groups/jpeg2000.html#getjp2image>`__.
+
+    Attributes
+    ----------
+    date : datetime
+        Desired date/time of the JP2 image.
+    sourceId : int
+        Unique image datasource identifier.
+    jpip : bool, optional
+        Returns a JPIP URI instead of the binary data of the image if set to `True`, defaults to `False`.
+    json : bool, optional
+        Returns the JSON if set to `True`, defaults to `False`.
+    """
+
+    date: datetime
+    sourceId: int
+    jpip: bool = False
+    Json: bool = Field(False, alias="json")
+
+    @validator("date")
+    def convert_date_to_isoformat(cls, v):
+        """
+        Converts the date from a datetime object to a string in the ISO format.
+        """
+        return v.isoformat() + "Z"
+
+    def get_output_type(self):
+        """
+        Returns the output type of the API call.
+        """
+        if self.Json and self.jpip:
+            return OutputType.Json
+        elif not self.Json and self.jpip:
+            return OutputType.String
+        else:
+            return OutputType.Raw

hvpy/api_groups/jpeg2000/tests/test_get_jp2_image.py

@@ -0,0 +1,73 @@
+from datetime import datetime
+
+import pytest
+from pydantic import ValidationError
+
+from hvpy.api_groups.jpeg2000.get_jp2_image import getJP2ImageInputParameters
+from hvpy.core import execute_api_call
+
+
+def test_str_response():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": True, "json": False}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, str)
+    assert response.startswith("jpip://")
+
+
+def test_json_response():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": True, "json": True}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, dict)
+    assert "uri" in response
+    assert response["uri"].startswith("jpip://")
+
+
+def test_raw_response():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": False, "json": False}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, bytes)
+
+
+def test_raw_response_with_json():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": False, "json": True}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, bytes)
+
+
+def test_default_response():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, bytes)
+
+
+def test_error_handling():
+    params = {"sourceId": 14, "jpip": True, "json": True}
+    with pytest.raises(ValidationError, match="getJP2ImageInputParameters\ndate\n  field required"):
+        getJP2ImageInputParameters(**params)
+
+
+def test_unknown_parameters():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": True, "json": True, "should_reject_this": True}
+    params = getJP2ImageInputParameters(**params)
+    response = execute_api_call(input_parameters=params)
+    assert isinstance(response, dict)
+    assert "uri" in response
+    assert response["uri"].startswith("jpip://")
+
+
+def test_url_property():
+    date_obj = datetime(2022, 1, 1, 23, 59, 59)
+    params = {"date": date_obj, "sourceId": 14, "jpip": True, "json": True}
+    params = getJP2ImageInputParameters(**params)
+    assert params.url == "https://api.helioviewer.org/v2/getJP2Image/"

hvpy/core.py

@@ -0,0 +1,47 @@
+import requests
+
+from hvpy.io import HvpyParameters, OutputType
+
+
+def parse_response(response: requests.Response, output_parameters: OutputType):
+    """
+    Parses the response from the API call based on the output type.
+
+    Parameters
+    ----------
+    response : requests.Response
+        The response from the API call.
+    output_parameters : OutputType
+        The output type.
+
+    Returns
+    -------
+    binary | str | json
+        The parsed response.
+    """
+
+    if output_parameters == OutputType.Raw:
+        return response.content
+    elif output_parameters == OutputType.String:
+        return response.content.decode("utf-8")
+    elif output_parameters == OutputType.Json:
+        return response.json()
+
+
+def execute_api_call(input_parameters: HvpyParameters):
+    """
+    Executes the API call and returns a parsed response.
+
+    Parameters
+    ----------
+    input_parameters : HvpyParameters
+        The input parameters.
+
+    Returns
+    -------
+    Parsed response from the API.
+    """
+
+    response = requests.get(input_parameters.url, params=input_parameters.dict())
+    response.raise_for_status()
+    return parse_response(response, input_parameters.get_output_type())

hvpy/io.py

@@ -0,0 +1,31 @@
+from enum import Enum, auto
+
+from pydantic import BaseModel
+
+BASE_URL = "https://api.helioviewer.org/v2/"
+
+
+class HvpyParameters(BaseModel):
+    def dict(self):
+        """
+        Pydantic doesn't allow using lowercase 'json' as a field, so we
+        override it.
+        """
+        d = super().dict()
+        if "Json" in d:
+            d["json"] = d["Json"]
+            del d["Json"]
+        return d
+
+    def get_output_type(self):
+        return OutputType.Raw
+
+    @property
+    def url(self):
+        return BASE_URL + self.__class__.__name__[:-15] + "/"
+
+
+class OutputType(Enum):
+    Raw = auto()
+    String = auto()
+    Json = auto()

hvpy/tests/test_io.py

@@ -0,0 +1,11 @@
+from hvpy.io import HvpyParameters, OutputType
+
+
+def test_default_get_output_type_is_raw():
+    params = HvpyParameters()
+    assert params.get_output_type() == OutputType.Raw
+
+
+def test_url_property():
+    params = HvpyParameters()
+    assert params.url == "https://api.helioviewer.org/v2//"

setup.cfg

@@ -33,6 +33,7 @@ setup_requires =
     setuptools_scm
 install_requires =
     requests>=2.27.0
+    pydantic>=1.9.1
 
 [options.extras_require]
 tests =