diff --git a/src/pysdmx/model/message.py b/src/pysdmx/model/message.py index d9daf33..a6a8402 100644 --- a/src/pysdmx/model/message.py +++ b/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) diff --git a/tests/model/test_message.py b/tests/model/test_message.py new file mode 100644 index 0000000..41f690d --- /dev/null +++ b/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)