Source code for sdmx.message
"""Classes for SDMX messages.
:class:`Message` and related classes are not defined in the SDMX
:doc:`information model <implementation>`, but in the
:ref:`SDMX-ML standard <formats>`.
:mod:`sdmx` also uses :class:`DataMessage` to encapsulate SDMX-JSON data
returned by data sources.
"""
from typing import List, Optional, Text, Union
from requests import Response
from sdmx import model
from sdmx.util import BaseModel, DictLike, summarize_dictlike
def _summarize(obj, fields):
"""Helper method for __repr__ on Header and Message (sub)classes."""
for name in fields:
attr = getattr(obj, name)
if attr is None:
continue
yield f"{name}: {repr(attr)}"
[docs]class Message(BaseModel):
[docs] class Config:
# for .response
arbitrary_types_allowed = True
#: :class:`Header` instance.
header: Header = Header()
#: (optional) :class:`Footer` instance.
footer: Optional[Footer] = None
#: :class:`requests.Response` instance for the response to the HTTP request
#: that returned the Message. This is not part of the SDMX standard.
response: Optional[Response] = None
def __str__(self):
return repr(self)
def __repr__(self):
"""String representation."""
lines = [
f"<sdmx.{self.__class__.__name__}>",
repr(self.header).replace("\n", "\n "),
]
lines.extend(_summarize(self, ["footer", "response"]))
return "\n ".join(lines)
[docs]class ErrorMessage(Message):
pass
[docs]class StructureMessage(Message):
#: Collection of :class:`.CategoryScheme`.
category_scheme: DictLike[str, model.CategoryScheme] = DictLike()
#: Collection of :class:`.Codelist`.
codelist: DictLike[str, model.Codelist] = DictLike()
#: Collection of :class:`.ConceptScheme`.
concept_scheme: DictLike[str, model.ConceptScheme] = DictLike()
#: Collection of :class:`.ContentConstraint`.
constraint: DictLike[str, model.ContentConstraint] = DictLike()
#: Collection of :class:`.DataflowDefinition`.
dataflow: DictLike[str, model.DataflowDefinition] = DictLike()
#: Collection of :class:`.DataStructureDefinition`.
structure: DictLike[str, model.DataStructureDefinition] = DictLike()
#: Collection of :class:`.AgencyScheme`.
organisation_scheme: DictLike[str, model.AgencyScheme] = DictLike()
#: Collection of :class:`.ProvisionAgreement`.
provisionagreement: DictLike[str, model.ProvisionAgreement] = DictLike()
def __repr__(self):
"""String representation."""
lines = [super().__repr__()]
# StructureMessage contents
for attr in self.__dict__.values():
if isinstance(attr, DictLike) and attr:
lines.append(summarize_dictlike(attr))
return "\n ".join(lines)
[docs]class DataMessage(Message):
"""Data Message.
.. note:: A DataMessage may contain zero or more :class:`.DataSet`, so
:attr:`data` is a list. To retrieve the first (and possibly only)
data set in the message, access the first element of the list:
``msg.data[0]``.
"""
#: :class:`list` of :class:`.DataSet`.
data: List[model.DataSet] = []
#: :class:`.DataflowDefinition` that contains the data.
dataflow: model.DataflowDefinition = model.DataflowDefinition()
#: The "dimension at observation level".
observation_dimension: Optional[
Union[
model._AllDimensions,
model.DimensionComponent,
List[model.DimensionComponent],
]
] = None
# Convenience access
@property
def structure(self):
"""DataStructureDefinition used in the :attr:`dataflow`."""
return self.dataflow.structure
def __repr__(self):
"""String representation."""
lines = [super().__repr__()]
# DataMessage contents
if self.data:
lines.append("DataSet ({})".format(len(self.data)))
lines.extend(_summarize(self, ("dataflow", "observation_dimension")))
return "\n ".join(lines)