Add builder for schema queries

src/pysdmx/api/qb/__init__.py

@@ -1,5 +1,6 @@
 """Build SDMX-REST queries."""
 
+from pysdmx.api.qb.schema import SchemaContext, SchemaFormat, SchemaQuery
 from pysdmx.api.qb.structure import (
     StructureDetail,
     StructureFormat,
@@ -11,6 +12,9 @@
 
 __all__ = [
     "ApiVersion",
+    "SchemaContext",
+    "SchemaFormat",
+    "SchemaQuery",
     "StructureDetail",
     "StructureFormat",
     "StructureQuery",

src/pysdmx/api/qb/schema.py

@@ -0,0 +1,154 @@
+"""Build SDMX-REST schema queries."""
+
+from enum import Enum
+from typing import Optional
+
+import msgspec
+
+from pysdmx.api.qb.structure import _V2_0_ADDED, StructureType
+from pysdmx.api.qb.util import ApiVersion, REST_ALL, REST_LATEST
+from pysdmx.errors import ClientError
+
+
+class SchemaContext(Enum):
+    """The context for which a schema must be generated."""
+
+    DATA_STRUCTURE = "datastructure"
+    METADATA_STRUCTURE = "metadatastructure"
+    DATAFLOW = "dataflow"
+    METADATA_FLOW = "metadataflow"
+    PROVISION_AGREEMENT = "provisionagreement"
+    METADATA_PROVISION_AGREEMENT = "metadataprovisionagreement"
+
+
+class SchemaFormat(Enum):
+    """The response formats."""
+
+    SDMX_JSON_1_0_0_SCHEMA = "application/vnd.sdmx.schema+json;version=1.0.0"
+    SDMX_JSON_2_0_0_SCHEMA = "application/vnd.sdmx.schema+json;version=2.0.0"
+    SDMX_JSON_1_0_0_STRUCTURE = (
+        "application/vnd.sdmx.structure+json;version=1.0.0"
+    )
+    SDMX_JSON_2_0_0_STRUCTURE = (
+        "application/vnd.sdmx.structure+json;version=2.0.0"
+    )
+    SDMX_ML_2_1_SCHEMA = "application/vnd.sdmx.schema+xml;version=2.1"
+    SDMX_ML_3_0_SCHEMA = "application/vnd.sdmx.schema+xml;version=3.0.0"
+    SDMX_ML_2_1_STRUCTURE = "application/vnd.sdmx.structure+xml;version=2.1"
+    SDMX_ML_3_0_STRUCTURE = "application/vnd.sdmx.structure+xml;version=3.0.0"
+
+
+class SchemaQuery(msgspec.Struct, frozen=True, omit_defaults=True):
+    """A schema query.
+
+    Schema queries allow retrieving the definition of data validity for a
+    certain context. The service must take into account the constraints
+    that apply within that context (e.g. dataflow).
+
+    Attributes:
+        context: The context for which a schema must be generated. This
+            determines the constraints that will be taken into
+            consideration.
+        agency_id: The agency maintaining the context to be considered.
+        resource_id: The id of the context to be considered.
+        version: The version of the context to be considered.
+        obs_dimension: The ID of the dimension at the observation level.
+    """
+
+    context: SchemaContext
+    agency_id: str
+    resource_id: str
+    version: str = REST_LATEST
+    obs_dimension: Optional[str] = None
+    explicit: bool = False
+
+    def validate(self) -> None:
+        """Validate the query."""
+        try:
+            decoder.decode(encoder.encode(self))
+        except msgspec.DecodeError as err:
+            raise ClientError(422, "Invalid Schema Query", str(err)) from err
+
+    def get_url(self, version: ApiVersion, omit_defaults: bool = False) -> str:
+        """The URL for the query in the selected SDMX-REST API version."""
+        self.__validate_query(version)
+        if omit_defaults:
+            return self.__create_short_query(version)
+        else:
+            return self.__create_full_query(version)
+
+    def __to_kw(self, val: str, ver: ApiVersion) -> str:
+        if val == "~" and ver < ApiVersion.V2_0_0:
+            val = "latest"
+        return val
+
+    def __check_context(self, version: ApiVersion) -> None:
+        ct = StructureType(self.context.value)
+        if version < ApiVersion.V2_0_0 and ct in _V2_0_ADDED:
+            raise ClientError(
+                422,
+                "Validation Error",
+                f"{self.context} not allowed in {version.value}.",
+            )
+
+    def __check_version(self) -> None:
+        if self.version == REST_ALL:
+            raise ClientError(
+                422,
+                "Validation Error",
+                "Retrieving schemas for all versions is not allowed.",
+            )
+
+    def __check_explicit(self, version: ApiVersion) -> None:
+        if self.explicit and version >= ApiVersion.V2_0_0:
+            raise ClientError(
+                422,
+                "Validation Error",
+                f"Explicit parameter is not supported in {version.value}.",
+            )
+
+    def __validate_query(self, version: ApiVersion) -> None:
+        self.validate()
+        self.__check_context(version)
+        self.__check_version()
+        self.__check_explicit(version)
+
+    def __create_full_query(self, ver: ApiVersion) -> str:
+        u = (
+            f"/schema/{self.context.value}/"
+            f"{self.agency_id}/{self.resource_id}"
+        )
+        u += f"/{self.__to_kw(self.version, ver)}"
+        if self.obs_dimension or ver < ApiVersion.V2_0_0:
+            u += "?"
+        if self.obs_dimension:
+            u += f"dimensionAtObservation={self.obs_dimension}"
+        if ver < ApiVersion.V2_0_0:
+            if self.obs_dimension:
+                u += "&"
+            u += f"explicit={str(self.explicit).lower()}"
+        return u
+
+    def __create_short_query(self, ver: ApiVersion) -> str:
+        u = (
+            f"/schema/{self.context.value}/"
+            f"{self.agency_id}/{self.resource_id}"
+        )
+        if self.version != REST_LATEST:
+            u += f"/{self.__to_kw(self.version, ver)}"
+        if self.obs_dimension or self.explicit:
+            u += "?"
+        if self.obs_dimension:
+            u += f"dimensionAtObservation={self.obs_dimension}"
+        if self.explicit:
+            if self.obs_dimension:
+                u += "&"
+            u += f"explicit={str(self.explicit).lower()}"
+        return u
+
+
+decoder = msgspec.json.Decoder(SchemaQuery)
+encoder = msgspec.json.Encoder()
+
+
+__all__ = ["SchemaContext", "SchemaFormat", "SchemaQuery"]

tests/api/qb/schema/test_schema_context.py

@@ -0,0 +1,16 @@
+from pysdmx.api.qb.schema import SchemaContext
+
+
+def test_expected_contexts():
+    expected = [
+        "datastructure",
+        "metadatastructure",
+        "dataflow",
+        "metadataflow",
+        "provisionagreement",
+        "metadataprovisionagreement",
+    ]
+
+    assert len(SchemaContext) == len(expected)
+    for fmt in SchemaContext:
+        assert fmt.value in expected

tests/api/qb/schema/test_schema_format.py

@@ -0,0 +1,18 @@
+from pysdmx.api.qb.schema import SchemaFormat
+
+
+def test_expected_formats():
+    expected = [
+        "application/vnd.sdmx.schema+json;version=1.0.0",
+        "application/vnd.sdmx.schema+xml;version=2.1",
+        "application/vnd.sdmx.schema+json;version=2.0.0",
+        "application/vnd.sdmx.schema+xml;version=3.0.0",
+        "application/vnd.sdmx.structure+xml;version=2.1",
+        "application/vnd.sdmx.structure+json;version=1.0.0",
+        "application/vnd.sdmx.structure+xml;version=3.0.0",
+        "application/vnd.sdmx.structure+json;version=2.0.0",
+    ]
+
+    assert len(SchemaFormat) == len(expected)
+    for fmt in SchemaFormat:
+        assert fmt.value in expected

tests/api/qb/schema/test_schema_query_context.py

@@ -0,0 +1,121 @@
+import pytest
+
+from pysdmx.api.qb.schema import (
+    SchemaContext,
+    SchemaQuery,
+)
+from pysdmx.api.qb.util import ApiVersion
+from pysdmx.errors import ClientError
+
+context_initial = [
+    SchemaContext.DATA_STRUCTURE,
+    SchemaContext.DATAFLOW,
+    SchemaContext.METADATA_STRUCTURE,
+    SchemaContext.METADATA_FLOW,
+    SchemaContext.PROVISION_AGREEMENT,
+]
+
+context_2_0_0 = [SchemaContext.METADATA_PROVISION_AGREEMENT]
+all_context = context_initial.copy()
+all_context.extend(context_2_0_0)
+
+
+@pytest.fixture()
+def agency():
+    return "BIS"
+
+
+@pytest.fixture()
+def res():
+    return "CBS"
+
+
+@pytest.fixture()
+def version():
+    return "1.0"
+
+
+@pytest.mark.parametrize(
+    "api_version", (v for v in ApiVersion if v < ApiVersion.V2_0_0)
+)
+@pytest.mark.parametrize("context", context_initial)
+def test_url_core_context_before_2_0_0(
+    context: SchemaContext,
+    agency: str,
+    res: str,
+    version: str,
+    api_version: ApiVersion,
+):
+    expected = (
+        f"/schema/{context.value}/{agency}/{res}/{version}?explicit=false"
+    )
+
+    q = SchemaQuery(context, agency, res, version)
+    url = q.get_url(api_version)
+
+    assert url == expected
+
+
+@pytest.mark.parametrize(
+    "api_version", (v for v in ApiVersion if v >= ApiVersion.V2_0_0)
+)
+@pytest.mark.parametrize("context", context_initial)
+def test_url_core_context_since_2_0_0(
+    context: SchemaContext,
+    agency: str,
+    res: str,
+    version: str,
+    api_version: ApiVersion,
+):
+    expected = f"/schema/{context.value}/{agency}/{res}/{version}"
+
+    q = SchemaQuery(context, agency, res, version)
+    url = q.get_url(api_version)
+
+    assert url == expected
+
+
+@pytest.mark.parametrize(
+    "api_version", (v for v in ApiVersion if v < ApiVersion.V2_0_0)
+)
+@pytest.mark.parametrize("context", context_2_0_0)
+def test_url_2_0_0_context_before_2_0_0(
+    context: SchemaContext,
+    agency: str,
+    res: str,
+    version: str,
+    api_version: ApiVersion,
+):
+    q = SchemaQuery(context, agency, res, version)
+
+    with pytest.raises(ClientError):
+        q.get_url(api_version)
+
+
+@pytest.mark.parametrize(
+    "api_version", (v for v in ApiVersion if v >= ApiVersion.V2_0_0)
+)
+@pytest.mark.parametrize("context", context_2_0_0)
+def test_url_2_0_0_context_since_2_0_0(
+    context: SchemaContext,
+    agency: str,
+    res: str,
+    version: str,
+    api_version: ApiVersion,
+):
+    expected = f"/schema/{context.value}/{agency}/{res}/{version}"
+
+    q = SchemaQuery(context, agency, res, version)
+    url = q.get_url(api_version)
+
+    assert url == expected
+
+
+def test_wrong_context(
+    agency: str,
+    res: str,
+):
+    sq = SchemaQuery(42, agency, res)
+
+    with pytest.raises(ClientError):
+        sq.get_url(ApiVersion.V1_0_0)