Skip to content
Back to top

# Source code for prov.serializers

    from __future__ import annotations  # needed for | type annotations in Python < 3.10

    import io
    from abc import ABC, abstractmethod
    from typing import TYPE_CHECKING, Any, ClassVar

    from prov import Error

    if TYPE_CHECKING:
        from prov.model import ProvDocument

    __author__ = "Trung Dong Huynh"
    __email__ = "trungdong@donggiang.com"

    __all__ = ["Registry", "Serializer", "get"]

    def _is_text_stream(stream: io.IOBase) -> bool:
        """Return whether ``stream`` accepts/produces ``str`` rather than bytes.

        ``isinstance(stream, io.TextIOBase)`` alone misses file-like wrappers
        such as ``tempfile.NamedTemporaryFile``'s ``_TemporaryFileWrapper``,
        which proxy a text stream without subclassing ``io.TextIOBase``. Such
        wrappers expose the underlying text stream's ``encoding`` attribute,
        while binary streams (``io.BytesIO``, buffered readers/writers) have no
        ``encoding``, so the extra check only ever reclassifies text-stream
        proxies that would otherwise crash on bytes (#240).
        """
        return isinstance(stream, io.TextIOBase) or hasattr(stream, "encoding")

    class Serializer(ABC):
        """Serializer for PROV documents."""

        document = None
        """PROV document to serialise."""

        def __init__(self, document: ProvDocument | None = None):
            """Create a serializer bound to a document.

            Args:
                document: Document this serializer will serialize, or
                    deserialize into. May be ``None`` when only deserialization
                    (which produces its own new document) is needed.
            """
            self.document = document

        @abstractmethod
        def serialize(self, stream: io.IOBase, **args: Any) -> None:
            """Serialize ``self.document`` and write it to a stream.

            Subclasses implement this to encode the bound document and write the
            result to ``stream``.

            Args:
                stream: Stream to write the serialized document into.
                **args: Format-specific serialization options, passed through by
                    subclasses.
            """
            pass  # pragma: no cover -- abstract body, never executed directly

        @abstractmethod
        def deserialize(self, stream: io.IOBase, **args: Any) -> ProvDocument:
            """Read and parse a document from a stream.

            Subclasses implement this to parse ``stream`` and build a new
            :class:`~prov.model.ProvDocument` from it.

            Args:
                stream: Stream to deserialize the document from.
                **args: Format-specific deserialization options, passed through
                    by subclasses.

            Returns:
                The deserialized :class:`~prov.model.ProvDocument`.
            """
            pass  # pragma: no cover -- abstract body, never executed directly

    class DoNotExist(Error):
        """Exception for the case a serializer is not available."""

        pass

    class Registry:
        """Registry of serializers."""

        serializers: ClassVar[dict[str, type[Serializer]] | None] = None
        """Property caching all available serializers in a dict."""

        @staticmethod
        def load_serializers() -> None:
            """Load all serializers whose optional dependencies are installed.

            ``json`` and ``provn`` have no optional dependencies and are always
            registered. ``rdf`` (needs ``rdflib``) and ``xml`` (needs ``lxml``)
            are registered only if their extra is installed, so that ``import
            prov`` and JSON/PROV-N work in a minimal install; requesting an
            unavailable format then raises an informative :class:`DoNotExist`
            (see :func:`get`) rather than a bare ``ModuleNotFoundError``.

            The insertion order is kept as ``json, rdf, provn, xml`` (the
            historic order when all extras are present) because :func:`prov.read`'s
            format auto-detection iterates ``Registry.serializers`` in order and
            ``test_read_auto_detect_with_broken_tell_degrades_to_no_rewind``
            pins JSON as the first candidate tried — a non-seekable stream can
            only attempt the first format, so JSON must remain first for JSON
            content to be auto-detected on broken streams.
            """
            from prov.serializers.provjson import ProvJSONSerializer
            from prov.serializers.provn import ProvNSerializer

            serializers: dict[str, type[Serializer]] = {
                "json": ProvJSONSerializer,
            }
            try:
                from prov.serializers.provrdf import ProvRDFSerializer
            except ImportError:  # pragma: no cover -- rdflib (rdf extra) absent; covered by the minimal-install CI job
                pass
            else:
                serializers["rdf"] = ProvRDFSerializer
            serializers["provn"] = ProvNSerializer
            try:
                from prov.serializers.provxml import ProvXMLSerializer
            except ImportError:  # pragma: no cover -- lxml (xml extra) absent; covered by the minimal-install CI job
                pass
            else:
                serializers["xml"] = ProvXMLSerializer
            Registry.serializers = serializers

    #: Formats provided by optional extras, used to build the DoNotExist message
    #: in :func:`get` when the corresponding dependency is not installed.
    _OPTIONAL_FORMAT_EXTRAS = {"rdf": "rdf", "xml": "xml"}

    def get(format_name: str) -> type[Serializer]:
        """Return the serializer class registered for a format.

        Args:
            format_name: Registry key, e.g. ``"json"``, ``"xml"``, ``"rdf"``,
                ``"provn"``.

        Returns:
            The :class:`Serializer` subclass for the format.

        Raises:
            DoNotExist: If no serializer is registered under ``format_name``.
        """
        # Lazily initialize the list of serializers to avoid cyclic imports
        if Registry.serializers is None:
            Registry.load_serializers()
        serializers = Registry.serializers
        assert serializers is not None  # load_serializers() always populates it
        try:
            return serializers[format_name]
        except KeyError as e:
            extra = _OPTIONAL_FORMAT_EXTRAS.get(format_name)
            # The informative-message branch is reachable only when the optional
            # extra is absent; the minimal-install CI job exercises it.
            if extra is not None:  # pragma: no cover
                raise DoNotExist(
                    f'Serializer for the "{format_name}" format requires the '
                    f'"{extra}" extra; install it with: pip install "prov[{extra}]"'
                ) from e
            raise DoNotExist(
                f'No serializer available for the format "{format_name}"'
            ) from e