Merge branch 'develop' into 44-read-xml-parsers-and-add-lxml-dependency

src/pysdmx/model/message.py

@@ -6,11 +6,15 @@
 
 from datetime import datetime, timezone
 from enum import Enum
-from typing import Optional
+from typing import Any, Dict, Optional
 import uuid
 
 from msgspec import Struct
 
+from pysdmx.errors import ClientError, NotFound
+from pysdmx.model import Codelist, ConceptScheme
+from pysdmx.model.__base import ItemScheme
+
 
 class ActionType(Enum):
     """ActionType enumeration.
@@ -34,3 +38,99 @@ class Header(Struct, frozen=True, kw_only=True):
     receiver: Optional[str] = None
     source: Optional[str] = None
     dataset_action: Optional[ActionType] = None
+
+
+ORGS = "OrganisationSchemes"
+CLS = "Codelists"
+CONCEPTS = "ConceptSchemes"
+
+MSG_CONTENT_PKG = {
+    ORGS: ItemScheme,
+    CLS: Codelist,
+    CONCEPTS: ConceptScheme,
+}
+
+
+class Message(Struct, frozen=True):
+    """Message class holds the content of SDMX Message.
+
+    Attributes:
+        content (Dict[str, Any]): Content of the message. The keys are the
+            content type (e.g. ``OrganisationSchemes``, ``Codelists``, etc.),
+            and the values are the content objects (e.g. ``ItemScheme``,
+            ``Codelist``, etc.).
+    """
+
+    content: Dict[str, Any]
+
+    def __post_init__(self) -> None:
+        """Checks if the content is valid."""
+        for content_key, content_value in self.content.items():
+            if content_key not in MSG_CONTENT_PKG:
+                raise ClientError(
+                    400,
+                    f"Invalid content type: {content_key}",
+                    "Check the docs for the proper structure on content.",
+                )
+
+            for obj_ in content_value.values():
+                if not isinstance(obj_, MSG_CONTENT_PKG[content_key]):
+                    raise ClientError(
+                        400,
+                        f"Invalid content value type: {type(obj_).__name__} "
+                        f"for {content_key}",
+                        "Check the docs for the proper "
+                        "structure on content.",
+                    )
+
+    def __get_elements(self, type_: str) -> Dict[str, Any]:
+        """Returns the elements from content."""
+        if type_ in self.content:
+            return self.content[type_]
+        raise NotFound(
+            404,
+            f"No {type_} found in content",
+            f"Could not find any {type_} in content.",
+        )
+
+    def __get_element_by_uid(self, type_: str, unique_id: str) -> Any:
+        """Returns a specific element from content."""
+        if type_ not in self.content:
+            raise NotFound(
+                404,
+                f"No {type_} found.",
+                f"Could not find any {type_} in content.",
+            )
+
+        if unique_id in self.content[type_]:
+            return self.content[type_][unique_id]
+
+        raise NotFound(
+            404,
+            f"No {type_} with id {unique_id} found in content",
+            "Could not find the requested element.",
+        )
+
+    def get_organisation_schemes(self) -> Dict[str, ItemScheme]:
+        """Returns the OrganisationScheme."""
+        return self.__get_elements(ORGS)
+
+    def get_codelists(self) -> Dict[str, Codelist]:
+        """Returns the Codelist."""
+        return self.__get_elements(CLS)
+
+    def get_concept_schemes(self) -> Dict[str, ConceptScheme]:
+        """Returns the Concept."""
+        return self.__get_elements(CONCEPTS)
+
+    def get_organisation_scheme_by_uid(self, unique_id: str) -> ItemScheme:
+        """Returns a specific OrganisationScheme."""
+        return self.__get_element_by_uid(ORGS, unique_id)
+
+    def get_codelist_by_uid(self, unique_id: str) -> Codelist:
+        """Returns a specific Codelist."""
+        return self.__get_element_by_uid(CLS, unique_id)
+
+    def get_concept_scheme_by_uid(self, unique_id: str) -> ConceptScheme:
+        """Returns a specific Concept."""
+        return self.__get_element_by_uid(CONCEPTS, unique_id)

tests/model/test_message.py

@@ -0,0 +1,127 @@
+import pytest
+
+from pysdmx.errors import ClientError, NotFound
+from pysdmx.model import Codelist, ConceptScheme
+from pysdmx.model.__base import ItemScheme
+from pysdmx.model.message import Message
+
+
+def test_initialization():
+    message = Message({})
+    assert message.content == {}
+
+
+def test_get_organisation():
+    org1 = ItemScheme(id="orgs1", agency="org1")
+    message = Message(
+        {
+            "OrganisationSchemes": {
+                "org1:orgs1(1.0)": org1,
+            }
+        }
+    )
+    assert message.get_organisation_schemes() == {
+        "org1:orgs1(1.0)": org1,
+    }
+
+    assert message.get_organisation_scheme_by_uid("org1:orgs1(1.0)") == org1
+
+
+def test_get_codelists():
+    cl1 = Codelist(id="cl1", agency="cl1")
+    message = Message(
+        {
+            "Codelists": {
+                "cl1:cl1(1.0)": cl1,
+            }
+        }
+    )
+    assert message.get_codelists() == {
+        "cl1:cl1(1.0)": cl1,
+    }
+
+    assert message.get_codelist_by_uid("cl1:cl1(1.0)") == cl1
+
+
+def test_get_concepts():
+    cs1 = ConceptScheme(id="cs1", agency="cs1")
+    message = Message(
+        {
+            "ConceptSchemes": {
+                "cs1:cs1(1.0)": cs1,
+            }
+        }
+    )
+    assert message.get_concept_schemes() == {
+        "cs1:cs1(1.0)": cs1,
+    }
+
+    assert message.get_concept_scheme_by_uid("cs1:cs1(1.0)") == cs1
+
+
+def test_empty_get_elements():
+    message = Message({})
+    with pytest.raises(NotFound) as exc_info:
+        message.get_organisation_schemes()
+
+    assert "No OrganisationSchemes found" in str(exc_info.value.title)
+
+    with pytest.raises(NotFound) as exc_info:
+        message.get_codelists()
+
+    assert "No Codelists found" in str(exc_info.value.title)
+
+    with pytest.raises(NotFound) as exc_info:
+        message.get_concept_schemes()
+
+    assert "No ConceptSchemes found" in str(exc_info.value.title)
+
+
+def test_empty_get_element_by_uid():
+    message = Message({})
+    with pytest.raises(NotFound) as exc_info:
+        message.get_organisation_scheme_by_uid("org1:orgs1(1.0)")
+
+    assert "No OrganisationSchemes found" in str(exc_info.value.title)
+
+    with pytest.raises(NotFound) as exc_info:
+        message.get_codelist_by_uid("cl1:cl1(1.0)")
+
+    assert "No Codelists found" in str(exc_info.value.title)
+
+    with pytest.raises(NotFound) as exc_info:
+        message.get_concept_scheme_by_uid("cs1:cs1(1.0)")
+
+    assert "No ConceptSchemes found" in str(exc_info.value.title)
+
+
+def test_invalid_get_element_by_uid():
+    message = Message({"OrganisationSchemes": {}})
+
+    e_m = "No OrganisationSchemes with id"
+
+    with pytest.raises(NotFound) as exc_info:
+        message.get_organisation_scheme_by_uid("org12:orgs1(1.0)")
+    assert e_m in str(exc_info.value.title)
+
+
+def test_invalid_initialization_content_key():
+    exc_message = "Invalid content type: Invalid"
+    with pytest.raises(ClientError) as exc_info:
+        Message({"Invalid": {}})
+    assert exc_message in str(exc_info.value.title)
+
+
+@pytest.mark.parametrize(
+    ("key", "value"),
+    [
+        ("OrganisationSchemes", {"org1:orgs1(1.0)": "invalid"}),
+        ("Codelists", {"cl1:cl1(1.0)": "invalid"}),
+        ("ConceptSchemes", {"cs1:cs1(1.0)": "invalid"}),
+    ],
+)
+def test_invalid_initialization_content_value(key, value):
+    exc_message = f"Invalid content value type: str for {key}"
+    with pytest.raises(ClientError) as exc_info:
+        Message({key: value})
+    assert exc_message in str(exc_info.value.title)