Hierarchy checks

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pysdmx"
-version = "1.0.0-beta-6"
+version = "1.0.0-beta-7"
 description = "Your opinionated Python SDMX library"
 authors = [
     "Xavier Sosnovsky <[email protected]>",

src/pysdmx/__init__.py

@@ -1,3 +1,3 @@
 """Your opinionated Python SDMX library."""
 
-__version__ = "1.0.0-beta-6"
+__version__ = "1.0.0-beta-7"

src/pysdmx/fmr/__init__.py

@@ -1,6 +1,14 @@
 """Retrieve metadata from an FMR instance."""
 from enum import Enum
-from typing import Any, Literal, NoReturn, Optional, Sequence, Tuple, Union
+from typing import (
+    Any,
+    Literal,
+    NoReturn,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+)
 
 import httpx
 from msgspec.json import decode
@@ -15,6 +23,7 @@
     ConceptScheme,
     DataflowInfo,
     Hierarchy,
+    HierarchyAssociation,
     MetadataReport,
     MultiRepresentationMap,
     Organisation,
@@ -70,6 +79,10 @@ class Context(Enum):
         "structure/dataflow/{0}/{1}/{2}"
         "?detail=referencepartial&references={3}"
     ),
+    "ha": (
+        "structure/dataflow/{0}/{1}/{2}"
+        "?references=all&detail=referencepartial"
+    ),
     "hierarchy": (
         "structure/hierarchy/{0}/{1}/{2}"
         "?detail=referencepartial&references=codelist"
@@ -217,6 +230,12 @@ def __fetch(self, url: str, is_ref_meta: bool = False) -> bytes:
             except (httpx.RequestError, httpx.HTTPStatusError) as e:
                 self._error(e)
 
+    def __get_hierarchies_for_flow(
+        self, agency: str, flow: str, version: str
+    ) -> Sequence[HierarchyAssociation]:
+        out = self.__fetch(super()._url("ha", agency, flow, version))
+        return super()._out(out, self.deser.hier_assoc)
+
     def get_agencies(self, agency: str) -> Sequence[Organisation]:
         """Get the list of **sub-agencies** for the supplied agency.
 
@@ -339,16 +358,14 @@ def get_schema(
         Returns:
             The requested schema.
         """
+        ha = (
+            self.__get_hierarchies_for_flow(agency, id, version)
+            if context != "datastructure"
+            else ()
+        )
         c = context.value if isinstance(context, Context) else context
         out = self.__fetch(super()._url("schema", c, agency, id, version))
-        return super()._out(
-            out,
-            self.deser.schema,
-            c,
-            agency,
-            id,
-            version,
-        )
+        return super()._out(out, self.deser.schema, c, agency, id, version, ha)
 
     def get_dataflow_details(
         self,
@@ -537,6 +554,12 @@ async def __fetch(self, url: str, is_ref_meta: bool = False) -> bytes:
             except (httpx.RequestError, httpx.HTTPStatusError) as e:
                 self._error(e)
 
+    async def __get_hierarchies_for_flow(
+        self, agency: str, flow: str, version: str
+    ) -> Sequence[HierarchyAssociation]:
+        out = await self.__fetch(super()._url("ha", agency, flow, version))
+        return super()._out(out, self.deser.hier_assoc)
+
     async def get_agencies(self, agency: str) -> Sequence[Organisation]:
         """Get the list of **sub-agencies** for the supplied agency.
 
@@ -657,16 +680,14 @@ async def get_schema(
         Returns:
             The requested schema.
         """
+        ha = (
+            await self.__get_hierarchies_for_flow(agency, id, version)
+            if context != "datastructure"
+            else ()
+        )
         c = context.value if isinstance(context, Context) else context
         r = await self.__fetch(super()._url("schema", c, agency, id, version))
-        return super()._out(
-            r,
-            self.deser.schema,
-            c,
-            agency,
-            id,
-            version,
-        )
+        return super()._out(r, self.deser.schema, c, agency, id, version, ha)
 
     async def get_dataflow_details(
         self,

src/pysdmx/fmr/fusion/__init__.py

@@ -3,6 +3,7 @@
 from pysdmx.fmr.fusion.category import FusionCategorySchemeMessage
 from pysdmx.fmr.fusion.code import (
     FusionCodelistMessage,
+    FusionHierarchyAssociationMessage,
     FusionHierarchyMessage,
 )
 from pysdmx.fmr.fusion.concept import FusionConcepSchemeMessage
@@ -24,6 +25,7 @@
     dataflow=FusionDataflowMessage,  # type: ignore[arg-type]
     providers=FusionProviderMessage,  # type: ignore[arg-type]
     schema=FusionSchemaMessage,  # type: ignore[arg-type]
+    hier_assoc=FusionHierarchyAssociationMessage,  # type: ignore[arg-type]
     hierarchy=FusionHierarchyMessage,  # type: ignore[arg-type]
     report=FusionMetadataMessage,  # type: ignore[arg-type]
     mapping=FusionMappingMessage,  # type: ignore[arg-type]

src/pysdmx/fmr/fusion/code.py

@@ -5,14 +5,15 @@
 
 from msgspec import Struct
 
-from pysdmx.fmr.fusion.core import FusionAnnotation, FusionString
+from pysdmx.fmr.fusion.core import FusionAnnotation, FusionLink, FusionString
 from pysdmx.model import (
     Code,
     Codelist as CL,
     HierarchicalCode,
     Hierarchy as HCL,
+    HierarchyAssociation as HA,
 )
-from pysdmx.util import parse_item_urn
+from pysdmx.util import find_by_urn, parse_item_urn
 
 
 class FusionCode(Struct, frozen=True):
@@ -54,6 +55,7 @@ class FusionCodelist(Struct, frozen=True, rename={"agency": "agencyId"}):
     """Fusion-JSON payload for a codelist."""
 
     id: str
+    urn: str
     names: Sequence[FusionString]
     agency: str
     descriptions: Sequence[FusionString] = ()
@@ -62,13 +64,15 @@ class FusionCodelist(Struct, frozen=True, rename={"agency": "agencyId"}):
 
     def to_model(self) -> CL:
         """Converts a JsonCodelist to a standard codelist."""
+        t = "codelist" if "Codelist" in self.urn else "valuelist"
         return CL(
             self.id,
             self.names[0].value,
             self.agency,
             self.descriptions[0].value if self.descriptions else None,
             self.version,
             [i.to_model() for i in self.items],
+            t,  # type: ignore[arg-type]
         )
 
 
@@ -155,6 +159,42 @@ def to_model(self, codelists: Sequence[CL]) -> HCL:
         )
 
 
+class FusionHierarchyAssociation(
+    Struct, frozen=True, rename={"agency": "agencyId"}
+):
+    """Fusion-JSON payload for a hierarchy association."""
+
+    id: str
+    names: Sequence[FusionString]
+    agency: str
+    hierarchyRef: str
+    linkedStructureRef: str
+    contextRef: str
+    links: Sequence[FusionLink] = ()
+    descriptions: Sequence[FusionString] = ()
+    version: str = "1.0"
+
+    def to_model(
+        self,
+        hierarchies: Sequence[FusionHierarchy],
+        codelists: Sequence[FusionCodelist],
+    ) -> HA:
+        """Converts a FusionHierarchyAssocation to a standard association."""
+        cls = [cl.to_model() for cl in codelists]
+        m = find_by_urn(hierarchies, self.hierarchyRef).to_model(cls)
+        return HA(
+            self.id,
+            self.names[0].value,
+            self.agency,
+            m,
+            self.linkedStructureRef,
+            self.contextRef,
+            self.descriptions[0].value if self.descriptions else None,
+            self.version,
+            self.links[0].urn if self.links else None,
+        )
+
+
 class FusionHierarchyMessage(Struct, frozen=True):
     """Fusion-JSON payload for /hierarchy queries."""
 
@@ -165,3 +205,18 @@ def to_model(self) -> HCL:
         """Returns the requested hierarchy."""
         cls = [cl.to_model() for cl in self.Codelist]
         return self.Hierarchy[0].to_model(cls)
+
+
+class FusionHierarchyAssociationMessage(Struct, frozen=True):
+    """Fusion-JSON payload for hierarchy associations."""
+
+    Codelist: Sequence[FusionCodelist] = ()
+    Hierarchy: Sequence[FusionHierarchy] = ()
+    HierarchyAssociation: Sequence[FusionHierarchyAssociation] = ()
+
+    def to_model(self) -> Sequence[HA]:
+        """Returns the requested hierarchy associations."""
+        return [
+            ha.to_model(self.Hierarchy, self.Codelist)
+            for ha in self.HierarchyAssociation
+        ]

src/pysdmx/fmr/fusion/concept.py

@@ -28,7 +28,7 @@ def to_model(self, codelists: Sequence[FusionCodelist]) -> Concept:
         c = (
             self.representation.to_enumeration(codelists, [])
             if self.representation
-            else []
+            else None
         )
         d = self.descriptions[0].value if self.descriptions else None
         cl_ref = (

src/pysdmx/fmr/fusion/core.py

@@ -3,27 +3,33 @@
 from datetime import datetime
 from typing import Any, Optional, Sequence, Union
 
-from msgspec import Struct
+import msgspec
 
-from pysdmx.model import ArrayBoundaries, Code, Facets
+from pysdmx.model import ArrayBoundaries, Codelist, Facets
 from pysdmx.util import find_by_urn
 
 
-class FusionAnnotation(Struct, frozen=True):
+class FusionAnnotation(msgspec.Struct, frozen=True):
     """Fusion-JSON payload for annotations."""
 
     title: str
     type: str
 
 
-class FusionString(Struct, frozen=True):
+class FusionString(msgspec.Struct, frozen=True):
     """Fusion-JSON payload for an international string."""
 
     locale: str
     value: str
 
 
-class FusionTextFormat(Struct, frozen=True):
+class FusionLink(msgspec.Struct, frozen=True):
+    """Fusion-JSON payload for link objects."""
+
+    urn: str
+
+
+class FusionTextFormat(msgspec.Struct, frozen=True):
     """Fusion-JSON payload for TextFormat."""
 
     textType: str
@@ -40,7 +46,7 @@ class FusionTextFormat(Struct, frozen=True):
     isSequence: bool = False
 
 
-class FusionRepresentation(Struct, frozen=True):
+class FusionRepresentation(msgspec.Struct, frozen=True):
     """Fusion-JSON payload for core representation."""
 
     textFormat: Optional[FusionTextFormat] = None
@@ -83,13 +89,16 @@ def to_enumeration(
         self,
         codelists: Sequence[Any],
         valid: Sequence[str],
-    ) -> Sequence[Code]:
+    ) -> Optional[Codelist]:
         """Returns the list of codes allowed for this component."""
-        codes = []
         if self.representation:
-            a = find_by_urn(codelists, self.representation).items
-            codes = [c.to_model() for c in a if not valid or c.id in valid]
-        return codes
+            a = find_by_urn(codelists, self.representation)
+            cl = a.to_model()
+            codes = [
+                c.to_model() for c in a.items if not valid or c.id in valid
+            ]
+            return msgspec.structs.replace(cl, codes=codes)
+        return None
 
     def to_array_def(self) -> Optional[ArrayBoundaries]:
         """Returns the array boundaries, if any."""