Source code for prov
from __future__ import annotations # needed for | type annotations in Python < 3.10
import io
import logging
import os
import warnings
from typing import IO, TYPE_CHECKING, Any, cast
if TYPE_CHECKING:
from prov.model import ProvDocument
__author__ = "Trung Dong Huynh"
__email__ = "trungdong@donggiang.com"
__version__ = "2.5.1"
__all__ = ["Error", "model", "read"]
[docs]
class Error(Exception):
"""Base class for all errors in this package."""
pass
[docs]
def read(
source: io.IOBase | IO[Any] | str | bytes | os.PathLike[str],
format: str | None = None,
) -> ProvDocument | None:
"""Read a :class:`~prov.model.ProvDocument` from a file, path, or string.
A ``str``/``bytes`` source naming an existing file is read from that
file; any other ``str``/``bytes`` is parsed as raw document content.
If ``format`` is not given, the format is detected by trying each
registered deserializer in turn and returning the first that both
succeeds and produces a non-empty document (a parse yielding no records
or bundles -- e.g. rdflib parsing empty/foreign input to an empty graph
-- is treated as "not this format"; registered namespaces are not used
as a signal here since the rdf deserializer always registers rdflib's
own default-bound namespace prefixes on every successful parse, empty
or not). Auto-detection swallows all deserializer errors; pass
``format`` explicitly to get the actual traceback from the matching
deserializer.
A seekable stream ``source`` (e.g. an open file object, ``io.StringIO``,
``io.BytesIO``) is rewound to its starting position before each
auto-detection attempt, so every candidate format sees the full content.
A non-seekable stream is consumed by the first candidate; pass
``format`` explicitly for those.
Args:
source: File-like stream, path to an existing file, or raw content.
format: Serialization format to use (e.g. ``"json"``, ``"xml"``,
``"rdf"``, ``"provn"``). If ``None``, every registered format is
tried in turn.
Returns:
The deserialized :class:`~prov.model.ProvDocument`.
Raises:
TypeError: If ``format`` is ``None`` and no registered serializer
produced a non-empty document from ``source``.
"""
# Lazy imports to not globber the namespace.
from prov.model import ProvDocument
from prov.serializers import Registry
Registry.load_serializers()
assert Registry.serializers is not None # populated by load_serializers()
serializers = Registry.serializers.keys()
src: io.IOBase | IO[Any] | str | bytes | os.PathLike[str] | None = source
content: str | bytes | None = None
if isinstance(src, (str, bytes)) and not os.path.isfile(src):
# Not a path to an existing file: treat the string itself as raw
# document content.
content, src = src, None
if format:
try:
return ProvDocument.deserialize(
source=src, content=content, format=format.lower()
)
except Exception:
if content is not None:
warnings.warn(
"read() source is not a path to an existing file and "
"was parsed as raw content; if you meant a file path, "
"check that it exists",
UserWarning,
stacklevel=2,
)
raise
# A stream source (as opposed to a path) is duck-typed on read(),
# matching ProvBundle.deserialize()'s own convention (#240). Only a
# seekable stream can be rewound between auto-detect attempts below;
# if the stream's own seekable()/tell() misbehave, degrade gracefully
# to the non-seekable behaviour (first candidate consumes the stream)
# rather than failing before any deserializer gets a chance.
stream: IO[Any] | None = (
cast(IO[Any], src) if src is not None and hasattr(src, "read") else None
)
start_pos: int | None = None
if stream is not None and hasattr(stream, "seekable"):
try:
if stream.seekable():
start_pos = stream.tell()
except Exception:
start_pos = None
# Failed-candidate diagnostics (e.g. rdflib's "does not look like a
# valid URI" logger warnings) are noise by definition here: every
# candidate's exception is swallowed below, so nothing about a failed
# attempt is ever surfaced to the caller anyway. Raise rdflib's logger
# level for the duration of auto-detection only -- the explicit-format
# path above is unaffected and still shows real diagnostics. Note this
# mutation is process-global and not thread-safe: two concurrent
# auto-detecting read() calls can race on the level, but the impact is
# bounded to transient log noise, so a Filter/thread-local would be
# more machinery than warranted.
rdflib_term_logger = logging.getLogger("rdflib.term")
previous_level = rdflib_term_logger.level
rdflib_term_logger.setLevel(logging.ERROR)
try:
for format in serializers:
if start_pos is not None:
assert stream is not None
try:
stream.seek(start_pos)
except Exception:
# A seek that fails mid-loop degrades to no-rewind for
# the remaining attempts rather than aborting detection.
start_pos = None
try:
document = ProvDocument.deserialize(
source=src, content=content, format=format
)
except Exception:
# Any failure from a candidate deserializer means "not this
# format" -- move on to the next candidate.
continue
if document.get_records() or document.has_bundles():
return document
# A parse producing a completely empty document (e.g. rdflib
# accepts empty input, or the xml deserializer walking a
# childless foreign root) is treated as not detected. Registered
# namespaces are deliberately not consulted: the rdf deserializer
# always copies rdflib's own default-bound namespace prefixes onto
# the document on every successful parse, so that signal is always
# truthy and would defeat this check for the rdf format.
finally:
rdflib_term_logger.setLevel(previous_level)
message = (
"Could not read from the source. To get a proper "
"error message, specify the format with the 'format' "
"parameter."
)
if content is not None:
message += (
" Note: the source is not a path to an existing file and was "
"parsed as raw content."
)
raise TypeError(message)