"""The public ``Schema`` object: one compiled XML Schema and its classes.
``Schema.compile`` runs the shared compilation pipeline
(:func:`_compile_into_context`) without an instance document:
fatal input problems raise :class:`~pyxsd.exceptions.PyXSDError`, while
schema-legality problems are collected on the schema's
:class:`~pyxsd.validation.ValidationReport` for
:meth:`Schema.require_valid` to surface.
"""
import decimal
import logging
import os.path
import warnings
import weakref
from collections.abc import Iterable, Iterator
from contextlib import contextmanager
from pathlib import Path
from typing import IO, TYPE_CHECKING, Any, Literal
from xml.etree import ElementTree as ET
from pyxsd.binding import ParseModes
from pyxsd.cli import _load_module_from_file
from pyxsd.document import Document
from pyxsd.dtd_declarations import (
DocumentDeclarations,
collect_document_declarations,
declaration_context,
)
from pyxsd.element_representatives.element_representative import (
ComponentTable,
ElementRepresentative,
componentKind,
)
from pyxsd.exceptions import PyXSDError, PyXSDWarning, ValidationError
from pyxsd.extensions import Extensions
from pyxsd.import_visibility import audit_direct_imports
from pyxsd.instance_binding import bind_instance
from pyxsd.locations import IssueLocation
from pyxsd.namespaces import XML_NS, NamespaceContext, parse_with_namespaces
from pyxsd.resource_policy import (
ResourcePolicy,
ResourcePolicyError,
ResourcePurpose,
new_validation_report,
read_resource,
record_declarations,
source_identifier,
validate_purpose,
)
from pyxsd.resource_refs import classify_resource, require_local_reference, resource_base
from pyxsd.schema_base import SchemaBase
from pyxsd.schema_checks import (
apply_default_attributes,
apply_default_open_content,
build_substitution_groups,
check_alternative_table_edc,
check_alternatives,
check_conditional_type_substitutable,
check_content_kind_derivation,
check_form_defaults,
check_group_redefine_restrictions,
check_keyref_references,
check_notation_uses,
check_simple_content_restriction_base,
check_substitution_group_exclusions,
check_type_references,
check_value_constraints,
report_declaration_issues,
report_open_content_derivations,
)
from pyxsd.schema_composition import (
CompositionContext,
apply_local_target_namespaces,
check_namespace_attribute_values,
collect_additional_schemas,
inject_xlink_namespace_attributes,
inject_xml_namespace_attributes,
inject_xsi_namespace_attributes,
refine_wildcard_specs,
schema_composition_context,
splice_additional_schemas,
splice_composed_schemas,
)
from pyxsd.schema_context import (
SchemaContext,
active_context,
context_provenance,
context_report,
remember_components,
)
from pyxsd.schema_hints import absolute_schema_location_pairs, resolve_schema_hint
from pyxsd.validation import ValidationReport
from pyxsd.version_gates import check_xsd10_vocabulary
from pyxsd.versioning import apply_conditional_inclusion
from pyxsd.xml_policy import XmlLimits, XmlParserPolicy
from pyxsd.xsd_data_types import ENTITIES, ENTITY
logger = logging.getLogger(__name__)
def _reaches_unparsed_entity_types(candidate: Any, visited: set[int]) -> bool:
"""Whether *candidate* is, or reaches, an ENTITY/ENTITIES datatype.
Reaches through inheritance (direct use, restrictions, simple-content
extensions), through a union's ``_unionMembers`` (the class's own or
an inherited member tuple), and through a schema list's ``itemType``.
*visited* makes the walk terminate on cycles that no XSD document
could express but a synthetic class graph can.
"""
if not isinstance(candidate, type):
return False
if id(candidate) in visited:
return False
visited.add(id(candidate))
if issubclass(candidate, (ENTITY, ENTITIES)):
return True
members = getattr(candidate, "_unionMembers", None)
if members is not None and any(
_reaches_unparsed_entity_types(member, visited) for member in members
):
return True
item = getattr(candidate, "itemType", None)
return item is not None and _reaches_unparsed_entity_types(item, visited)
def _schema_validates_unparsed_entities(schema: "Schema") -> bool:
"""Whether *schema* can validate any ENTITY/ENTITIES value.
Walks every generated class and every element/attribute descriptor
installed on one: a declaration typed directly by the built-in
``xs:ENTITY``/``xs:ENTITIES`` produces no generated datatype class of
its own, so the descriptors' resolved types are the entry points the
class walk would otherwise miss. A descriptor whose type cannot be
resolved cannot validate anything and is skipped. The built-in
datatype registry alone never activates anything.
"""
visited: set[int] = set()
candidates: list[Any] = []
for cls in schema.classes.values():
candidates.append(cls)
for descriptor in vars(cls).values():
# ``getType`` is an Element/Attribute declaration feature,
# not part of the shared representative base.
resolver = getattr(descriptor, "getType", None)
if resolver is None:
continue
try:
resolved = resolver()
except Exception:
continue
if resolved is not None:
candidates.append(resolved)
return any(_reaches_unparsed_entity_types(candidate, visited) for candidate in candidates)
def _stream_is_seekable(stream: Any) -> bool:
"""Whether *stream* can rewind; a stream without the method cannot."""
seekable = getattr(stream, "seekable", None)
return seekable is not None and seekable()
def _max_bytes(limits: XmlLimits | None) -> int | None:
"""The byte budget ``limits`` declares, or ``None`` for unlimited."""
return None if limits is None else limits.max_bytes
class _CollectionByteLimitReached(Exception):
"""Declaration collection crossed the caller's ``max_bytes`` budget."""
class _CollectDeclarations:
"""Sentinel: :meth:`Schema.parse` collects declaration evidence itself."""
_COLLECT_DECLARATIONS = _CollectDeclarations()
class _ByteCappedSource:
"""Read adapter enforcing ``xml_limits.max_bytes`` during collection.
The semantic parse enforces the byte budget through its own limited
reader; this adapter guarantees the declaration read -- which happens
beside it, before or after the parse -- can never stream past the same
budget. Crossing the cap aborts collection (the caller degrades the
evidence to ``None``) while the ordinary parser stays responsible for
the standard limit failure. Chunks are measured the way
``namespaces._ByteLimitedReader`` measures them: bytes by length,
text by its UTF-8 encoding.
"""
def __init__(self, stream: Any, max_bytes: int) -> None:
self._stream = stream
self._remaining = max_bytes
def read(self, size: int = -1) -> Any:
data = self._stream.read(size)
if data:
measured = len(data.encode("utf-8")) if isinstance(data, str) else len(data)
self._remaining -= measured
if self._remaining < 0:
raise _CollectionByteLimitReached("declaration read exceeded max_bytes")
return data
def _capped_for_collection(stream: Any, max_bytes: int | None) -> Any:
"""Wrap *stream* for collection, or pass it through without a budget."""
if max_bytes is None:
return stream
return _ByteCappedSource(stream, max_bytes)
class _PrologReplay:
"""A non-seekable stream that replays prolog bytes the collector read.
The declaration collector stops at the root start tag but consumes
stream data up to the end of that chunk. Before the semantic parse
runs, this reader serves those buffered bytes/str back first, then
chains the remaining stream -- the complete document stays bound and
nothing beyond the collector's chunk window is buffered.
"""
def __init__(self, stream: Any) -> None:
self._stream = stream
self._chunks: list[Any] = []
self._chunk_index = 0
self._offset = 0
self._collecting = True
def read(self, size: int = -1) -> Any:
if self._collecting:
data = self._stream.read(size)
if data:
self._chunks.append(data)
return data
return self._replay_read(size)
def finish_collection(self) -> None:
"""Switch from recording the collector's reads to replaying them."""
self._collecting = False
def seekable(self) -> bool:
return False
def _replay_read(self, size: int) -> Any:
parts: list[Any] = []
remaining: int | None = None if size < 0 else size
while self._chunk_index < len(self._chunks):
chunk = self._chunks[self._chunk_index]
piece = (
chunk[self._offset :]
if remaining is None
else chunk[self._offset : self._offset + remaining]
)
parts.append(piece)
self._offset += len(piece)
if remaining is not None:
remaining -= len(piece)
if self._offset >= len(chunk):
self._chunk_index += 1
self._offset = 0
if remaining is not None and remaining <= 0:
break
if self._chunk_index >= len(self._chunks):
tail = self._stream.read(-1 if remaining is None else remaining)
if tail:
parts.append(tail)
if not parts:
# Both the buffer and the stream are exhausted; hand the
# caller the same emptiness the underlying stream would.
return self._stream.read(0)
return parts[0][:0].join(parts)
if TYPE_CHECKING:
from pyxsd.batch import DocumentSource, ParseOutcome
from pyxsd.introspection import _Index
class _CompileHost:
"""Minimal parser-like host for ``Schema.compile``.
The compilation pipeline and the ElementRepresentative APIs read a
handful of attributes off the running parser (via ``host``/``parser``
parameters). A grep over the moved code (``pyxsd.schema_composition``,
``pyxsd.schema_checks``, ``pyxsd.content_model``, the
``element_representatives`` package and ``pyxsd.schema_base``)
enumerates exactly: ``report``, ``components``, ``mode``,
``classes``, ``namespaceContext`` and ``namespaceSchemas`` (the
latter only read by the pipeline body itself). Every container
aliases the corresponding :class:`CompositionContext` field, so the
products compiled through this host are the context's objects.
"""
components: Any
def __init__(self, ctx: CompositionContext) -> None:
self.report = ctx.report
self.classes = ctx.classes
self.mode = ctx.mode
self.processor_version = ctx.processor_version
self.extensions = ctx.extensions
self.namespaceContext = ctx.namespace_context
self.namespaceSchemas = ctx.namespace_schemas
# Replaced by the pipeline once the schema representative
# exists; ``None`` up to that point resolves lookups onto the
# module-level registry exactly as an un-run parser does.
self.components = None
[docs]
class Schema:
"""A compiled XML Schema and the classes generated from it."""
#: Generated classes keyed by declaration name (bare local names
#: and, where the type has an expanded name, Clark names).
classes: dict[str, type[SchemaBase]]
#: The component table registered during compilation.
components: Any
#: Prefix-to-URI bindings captured while parsing the schema documents.
namespace_context: Any
#: The binding policy the compilation ran under.
mode: Any
#: The XSD processor version this schema was compiled under ("1.0" or
#: "1.1"). It is the declared ``V`` that ``vc:*`` selectors were
#: tested against; the schema's own vocabulary is not restricted by it
#: (that is the compiler's job, see ``version_gates``).
xsd_version: str
#: The main schema document's ``targetNamespace``, if it declared one.
target_namespace: str | None
#: Schema-phase validation issues (composition, declaration, and
#: derivation problems). Instance-phase issues never land here.
report: ValidationReport
#: The source the schema was compiled from, as given to ``compile``.
source: str | Path | os.PathLike[str] | IO[str] | None
#: The parser-like host the compilation ran under (reached from a
#: generated class via its ``schema`` stamp). Per-parse binding
#: indexes attach here.
_host: Any
def __init__(
self,
classes: dict[str, type[SchemaBase]],
components: Any,
namespace_context: Any,
mode: Any,
target_namespace: str | None,
report: ValidationReport,
source: str | Path | os.PathLike[str] | IO[str] | None,
schema_context: SchemaContext | None = None,
) -> None:
self.classes = classes
self.components = components
self.namespace_context = namespace_context
self.mode = mode
self.target_namespace = target_namespace
self.report = report
self.source = source
self._schema_context = schema_context
self._host = None
# Lazily computed: whether any generated class can validate
# ENTITY/ENTITIES values. ``None`` means "not computed yet";
# ``load_overlay_classes`` resets it so late overlays re-activate.
self._entity_declarations_needed: bool | None = None
self.resource_policy: ResourcePolicy | None = None
self._introspection_index: _Index | None = None
self.declaration_origins: weakref.WeakKeyDictionary[ET.Element, IssueLocation] = (
weakref.WeakKeyDictionary()
)
[docs]
@classmethod
def compile(
cls,
source: str | Path | os.PathLike[str] | IO[str],
*,
mode: Any = ParseModes.STRICT,
xsd_version: Literal["1.0", "1.1"] = "1.1",
namespace_schemas: dict[str, str | Path] | None = None,
overlay: str | Path | os.PathLike[str] | None = None,
namespace_context: NamespaceContext | None = None,
schema_location_pairs: list[tuple[str | None, str]] | None = None,
extensions: Extensions | None = None,
resource_policy: ResourcePolicy | None = None,
xml_policy: XmlParserPolicy | None = None,
xml_limits: XmlLimits | None = None,
_source_purpose: ResourcePurpose = "schema",
) -> "Schema":
"""Compiles *source* into a :class:`Schema`.
- ``source`` - a schema file path or a file object open for
reading. An unopenable file, malformed XML, or a document
whose root is not ``xs:schema`` raise
:class:`~pyxsd.exceptions.PyXSDError`; schema-legality
problems are collected on :attr:`report` instead.
- ``mode`` - a :class:`~pyxsd.binding.BindingPolicy` (usually a
:class:`~pyxsd.binding.ParseModes` preset). Defaults to
:attr:`~pyxsd.binding.ParseModes.STRICT`.
- ``xsd_version`` - the XSD processor version the document is
compiled as, ``"1.0"`` or ``"1.1"`` (default ``"1.1"``). It is
the declared version ``vc:*`` conditional-inclusion selectors
test against: in ``"1.0"`` mode a declaration carrying
``vc:minVersion="1.1"`` is dropped (XSD 1.1 §4.2.2). The
vocabulary itself is checked separately (see
:mod:`pyxsd.version_gates`).
- ``namespace_schemas`` - an optional ``{namespace_uri: path}``
mapping supplying schemas for namespaces referenced by
``xs:import`` without a ``schemaLocation``.
- ``overlay`` - an optional overlay class file, loaded after
class generation (experimental; see
:func:`load_overlay_classes`).
- ``namespace_context`` - an existing prefix-to-URI binding
context for the schema documents' bindings to accumulate
into, instead of a fresh one (:func:`parse` passes the
context that already holds the instance's bindings, so one
context accumulates them all).
- ``schema_location_pairs`` - extra ``(namespace, location)``
pairs from an instance document's ``xsi:schemaLocation``
beyond the main schema. Advisory hints: consulted in strict
namespace mode only, and one that cannot be loaded is a
``schema-hint`` warning (:func:`parse` passes the instance's
pairs; a schema-only compile has none).
- ``extensions`` - an optional :class:`~pyxsd.extensions.
Extensions` registry selecting mixins to fold into the
generated classes' bases at build time (``XsdType.
getBaseList``). Without one, class generation is unchanged.
- ``resource_policy`` - a trusted application resource policy.
Its ``parse()`` owns XML parsing, so ``xml_policy`` and
``xml_limits`` cannot be combined with it.
- ``xml_policy`` - an opt-in :class:`~pyxsd.xml_policy.
XmlParserPolicy` applied to the main schema and every
included/imported dependency. ``None`` keeps stdlib parsing
and its current DTD/entity behavior. The choice is compile-
scoped: it never becomes the default for later
:meth:`parse` calls.
- ``xml_limits`` - optional :class:`~pyxsd.xml_policy.
XmlLimits` byte/depth/element budgets for the same documents.
- ``_source_purpose`` - internal: the resource purpose the main
schema document reads under. :func:`pyxsd.parse` passes
``"schema-hint"`` when the instance's own location hint
selected the schema and ``"schema"`` when the caller named it.
Use :meth:`require_valid` to reject a schema whose report
holds errors.
"""
if xsd_version not in ("1.0", "1.1"):
raise ValueError(f"xsd_version must be '1.0' or '1.1', not {xsd_version!r}")
purpose = validate_purpose(_source_purpose)
_reject_parser_override(resource_policy, xml_policy, xml_limits)
if overlay is not None and classify_resource(overlay).kind == "remote":
raise ResourcePolicyError(
f"the overlay '{overlay}' is remote; Python overlay modules are "
"never retrieved remotely"
)
if (
overlay is not None
and isinstance(source, (str, os.PathLike))
and classify_resource(source).kind == "remote"
):
raise ResourcePolicyError(
f"the remote schema '{source}' cannot load the local overlay "
f"'{overlay}'; Python overlays are never loaded for remote schemas"
)
with _policy_compilation(resource_policy, root_read=isinstance(source, (str, os.PathLike))):
original_source = source
source_data = None
if resource_policy is not None:
try:
source_data = read_resource(source, resource_policy, purpose=purpose)
except OSError as exc:
raise PyXSDError(f"the schema file could not be opened: {exc}") from exc
source = source_data.source_id
elif isinstance(source, (str, os.PathLike)):
require_local_reference(os.fspath(source), purpose=purpose)
main_source = resource_base(source) if isinstance(source, (str, os.PathLike)) else None
# A remote schema never loads a local Python overlay: that would be a
# remote-to-local pivot. The fallback directory (for an overlay named
# as a module) is derived only from a local source; a remote or
# memory source has no local directory to borrow.
overlay_dir: Path | None = None
if overlay is not None:
if main_source is not None and classify_resource(main_source).kind == "remote":
raise ResourcePolicyError(
f"the remote schema '{main_source}' cannot load the local overlay "
f"'{overlay}'; Python overlays are never loaded for remote schemas"
)
if main_source is None:
overlay_dir = Path.cwd()
elif classify_resource(main_source).kind == "local":
overlay_dir = Path(main_source).parent
ctx = CompositionContext(
mode=mode,
processor_version=decimal.Decimal(xsd_version),
namespace_schemas=dict(namespace_schemas or {}),
namespace_context=(
namespace_context if namespace_context is not None else NamespaceContext()
),
report=new_validation_report(resource_policy),
classes={},
main_source=main_source,
schema_context=SchemaContext(components=ComponentTable()),
namespace_overrides={},
component_namespaces={},
injected_builtin_ids=set(),
form_defaults={},
xpath_default_namespaces={},
composed_element_ids=set(),
composed_schema_roots={},
composed_schema_namespaces={},
directive_ids={},
compose_stack=[],
redefine_origins={},
override_origins={},
resolved_imports=set(),
composed_target_namespaces=set(),
extensions=extensions,
resource_policy=resource_policy,
source_data=source_data,
xml_policy=xml_policy,
xml_limits=xml_limits,
)
# Without an instance document there are no schemaLocation
# hints of our own; the caller (:func:`parse`) may forward the
# instance document's own pairs. Namespace-supplied schemas
# still resolve relative to the main schema document.
ctx.additional_schemas = collect_additional_schemas(ctx, schema_location_pairs or [])
host = _CompileHost(ctx)
schema = _compile_into_context(ctx, source, host)
schema.source = original_source
if overlay:
load_overlay_classes(schema, overlay, fallback_dir=overlay_dir)
return schema
[docs]
def require_valid(self) -> None:
"""Raises :class:`pyxsd.ValidationError` if the report has errors."""
if self.report.has_errors:
raise ValidationError(
f"the schema is not valid: {self.report.error_count} error(s); "
"inspect ValidationError.report",
self.report,
)
def _needs_entity_declarations(self) -> bool:
"""Whether this schema's classes can validate ENTITY/ENTITIES values.
Computed once per schema (after overlays may have been loaded) and
cached; a schema that cannot validate such values never runs a
declaration collection.
"""
if self._entity_declarations_needed is None:
self._entity_declarations_needed = _schema_validates_unparsed_entities(self)
return self._entity_declarations_needed
def _collect_root_declarations(
self, root: ET.Element, policy: XmlParserPolicy, max_bytes: int | None = None
) -> DocumentDeclarations | None:
"""Collect declaration evidence for an already-parsed *root*.
The re-read source comes from the root's recorded provenance (a
path to reopen, a seekable stream to rewind). This runs strictly
after a successful instance parse, so any failure to re-read --
the source vanished, a rewind broke, a policy denial, or the
re-read crossing ``max_bytes`` -- degrades to ``None``
(lexical-only) instead of turning the parse into a fatal read
error. Roots without provenance (caller-supplied
elements, non-seekable discovery streams) return ``None``.
"""
if not self._needs_entity_declarations():
return None
source = self.namespace_context.root_source(root)
if source is None:
return None
try:
if isinstance(source, str):
with open(source, "rb") as stream:
return collect_document_declarations(
_capped_for_collection(stream, max_bytes), policy=policy
)
marker = source.tell()
try:
source.seek(0)
return collect_document_declarations(
_capped_for_collection(source, max_bytes), policy=policy
)
finally:
source.seek(marker)
except Exception:
return None
[docs]
def stub(
self,
*,
compact: bool = False,
component_ids: tuple[str, ...] | None = None,
depth: int | None = None,
) -> str:
"""Returns ``.pyi``-style source for the classes this schema compiled.
The text is a typing surface mirroring the runtime: one class
block per generated class, annotated with the classes a valid
parsed document actually reads back, and the root elements as
module-level class aliases. A member whose type cannot be named
truthfully degrades to ``Any`` with the reason in a comment,
never to a name the runtime does not use.
- ``compact`` - emit the agent-oriented compact description
instead of the full document: one line per class and member,
a short alias legend, and the root elements in a trailing
comment. The compact text is a description, not valid Python;
the default full form is importable-quality ``.pyi`` source::
schema.stub() # full .pyi document
schema.stub(compact=True) # compact agent-oriented view
The CLI emits the same text:
``pyxsd --emit-stubs [PATH] -s SCHEMA`` writes the full
document to PATH (stdout when PATH is omitted), and adding
``--stubs-compact`` selects the compact view.
``component_ids`` optionally selects native declaration IDs, with
``depth`` bounding their outgoing dependencies (None follows all).
Unknown IDs raise KeyError and negative depths raise ValueError.
See :mod:`pyxsd.introspection` for IDs and declaration metadata.
Unfiltered emitting never raises for a schema that compiled and does not
require a valid report; call :meth:`require_valid` separately
to reject a schema whose report holds errors.
"""
from pyxsd.stubs import emit_stub
return emit_stub(self, compact=compact, component_ids=component_ids, depth=depth)
[docs]
def iter_parse(
self,
sources: Iterable["DocumentSource"],
*,
errors: Literal["raise", "report"] = "raise",
) -> Iterator["ParseOutcome"]:
"""Lazily parse explicit local sources using this compiled schema.
On first iteration, require a valid schema before consuming inputs.
Default raise mode stops with :class:`pyxsd.batch.BatchParseError`;
report mode yields invalid documents and expected input failures.
Source IDs must be unique. Internal defects always propagate.
Do not parse concurrently or reentrantly on this schema.
"""
from pyxsd.batch import iter_parse
return iter_parse(self, sources, errors=errors)
[docs]
def parse(
self,
xml: str | Path | os.PathLike[str] | IO[str] | ET.Element,
*,
xml_policy: XmlParserPolicy | None = None,
xml_limits: XmlLimits | None = None,
) -> "Document":
"""Binds *xml* against this schema and returns a :class:`Document`.
- ``xml`` - the instance document: a file path, a file object
open for reading, or a parsed ``xml.etree.ElementTree.Element``
used as the document root directly. An unopenable file or
malformed XML raises :class:`~pyxsd.exceptions.PyXSDError`
(the fatal-input rule); content problems instead land on the
document's report.
- ``xml_policy`` / ``xml_limits`` - opt-in parser posture and
structural budgets for this instance parse only. They do not
inherit a nondefault policy from :meth:`compile`; ``None``
keeps stdlib parsing and unlimited structure. They cannot be
combined with a custom ``resource_policy``, whose ``parse()``
is trusted application code. A pre-parsed ``Element`` is also
a caller-trusted boundary, so no parser option is applied.
Each parse binds with its own fresh instance report, so a
schema may serve many documents without their findings mixing:
the returned document's :attr:`~pyxsd.document.Document.report`
merges this schema's (frozen) compilation issues with the
document's own binding issues. ``Document.root`` is ``None``
when the document root matched no global element declaration
(the ``unknown-root`` issue records why).
"""
return self._parse_instance(xml, xml_policy=xml_policy, xml_limits=xml_limits)
def _parse_instance(
self,
xml: str | Path | os.PathLike[str] | IO[str] | ET.Element,
*,
xml_policy: XmlParserPolicy | None = None,
xml_limits: XmlLimits | None = None,
declarations: DocumentDeclarations | _CollectDeclarations | None = _COLLECT_DECLARATIONS,
) -> "Document":
"""One instance bind; *declarations* is internal revalidation evidence.
Omitted, this parse collects declaration evidence itself. A
:class:`~pyxsd.document.Document` revalidating its serialized tree
passes the evidence its original parse recorded -- including
``None`` -- because the serialized text has no DOCTYPE to
recollect from. ``None`` therefore means lexical-only, never an
empty, complete set.
"""
_reject_parser_override(self.resource_policy, xml_policy, xml_limits)
collection_policy = xml_policy if xml_policy is not None else XmlParserPolicy.default()
collected: DocumentDeclarations | None
if isinstance(declarations, _CollectDeclarations):
collected = None
collect_declarations = True
else:
collected = declarations
collect_declarations = False
if self.resource_policy is not None:
try:
data = read_resource(xml, self.resource_policy, purpose="instance")
tree = self.resource_policy.parse(data, self.namespace_context)
except OSError as exc:
raise PyXSDError(f"the xml input could not be read: {exc}") from exc
except ET.ParseError as exc:
raise PyXSDError(f"the xml file is not well-formed XML: {exc}") from exc
# A trusted policy parse is a caller boundary: no core-admitted
# bytes, so declaration evidence stays unavailable.
return self._bind_tree(tree, xml, source_id=data.source_id)
if isinstance(xml, ET.Element):
if collect_declarations:
collected = self._collect_root_declarations(
xml, collection_policy, _max_bytes(xml_limits)
)
return self._bind_tree(
xml,
xml,
source_id=source_identifier(xml),
declarations=collected,
)
original = xml
parse_source: Any = xml
if (
collect_declarations
and self._needs_entity_declarations()
and not isinstance(xml, (str, os.PathLike))
and not _stream_is_seekable(xml)
):
# A non-seekable stream cannot be reopened or rewound, so the
# collector runs first and the semantic parser then consumes
# the replayed prolog followed by the remaining stream. The
# collector never reads past the caller's byte budget: when
# the prolog alone crosses it, collection stops (evidence
# stays unavailable) and the semantic parser raises the
# standard limit failure over the replayed bytes.
replay = _PrologReplay(xml)
try:
collected = collect_document_declarations(
_capped_for_collection(replay, _max_bytes(xml_limits)),
policy=collection_policy,
)
except _CollectionByteLimitReached:
collected = None
finally:
replay.finish_collection()
parse_source = replay
logger.debug("The XML file is being parsed by the ElementTree library...")
try:
tree = parse_with_namespaces(
parse_source,
self.namespace_context,
policy=xml_policy,
limits=xml_limits,
)
except OSError as e:
raise PyXSDError(f"the xml input could not be read: {e}") from e
except ET.ParseError as e:
raise PyXSDError(f"the xml file is not well-formed XML: {e}") from e
logger.debug("XML file parsed by the ElementTree library successfully...")
if collect_declarations and collected is None:
collected = self._collect_root_declarations(
tree, collection_policy, _max_bytes(xml_limits)
)
return self._bind_tree(
tree,
original,
source_id=source_identifier(original),
declarations=collected,
)
def _bind_tree(
self,
tree: ET.Element,
source: Any,
*,
source_id: str | None,
declarations: DocumentDeclarations | None = None,
) -> "Document":
"""Bind an already verified tree; public policy inputs cannot use this path.
``declarations`` is this parse's DTD declaration evidence (or
``None`` for lexical-only ENTITY/ENTITIES validation); it is
installed around ``bind_instance`` only, never left behind.
"""
instance_report = new_validation_report(self.resource_policy)
from pyxsd.locations import ParseProvenance
provenance = ParseProvenance(tree, source_id)
context = self._schema_context
if context is None:
context = SchemaContext(components=self.components)
self._schema_context = context
# XSD 1.1 attribute inheritance and conditional type assignment
# both need to relate a bound element to its ancestors: the
# parent links and governing types are indexed once per parse,
# and the governing class of each bound element is recorded as
# binding proceeds. Binding code (schema_base) reads both back
# through the host the generated classes reach via their
# ``schema`` stamp.
#
# The dicts are freshly created for every parse, but they are
# attached to this schema's shared host, so parses against one
# ``Schema`` instance must not run concurrently -- a second
# parse would overwrite the indexes the first is still using.
# They are transient binding indexes, not document storage. Clear
# them even when binding raises so a cached Schema never retains
# the most recently parsed source tree through these parent links.
host = self._host
if host is not None:
host._elementParents = {}
host._elementTypes = {}
stack = [tree]
while stack:
parent = stack.pop()
for child in parent:
host._elementParents[id(child)] = parent
stack.append(child)
# Binding diagnostics route through the active context's report,
# so this parse's fresh report is installed (and restored) around
# the binding run; generated classes keep their ``cls.schema``
# stamp, whose report is the frozen compilation report.
try:
with (
active_context(context),
context_report(context, instance_report),
context_provenance(context, provenance),
declaration_context(declarations) as entity_evidence,
instance_report._at(provenance.source_location(tree)),
):
root = bind_instance(self, tree, instance_report)
provenance.bind(tree, root)
finally:
if host is not None:
host._elementParents.clear()
host._elementTypes.clear()
# The report-owning layer turns "unavailable external declarations
# could have declared a missing name" into exactly one advisory per
# parse, never one issue per suppressed value. It carries the
# document root's source address: the skipped external declaration
# has no occurrence of its own in the admitted bytes.
if entity_evidence is not None and entity_evidence.external_subset_skipped:
instance_report.add_warning(
"the document's external DTD declarations were not retrieved, so "
"ENTITY/ENTITIES values undeclared in the internal subset "
"could not be rejected",
code="dtd-external-skipped",
location=provenance.source_location(tree),
)
# One merged report per document: the schema phase's issues
# followed by this parse's instance issues. Issue objects are
# shared, so the phase tags recorded when each was found ride
# along, and the cached object's identity is stable for
# require_valid.
merged = new_validation_report(self.resource_policy)
merged.extend(self.report)
merged.extend(instance_report)
return Document(
schema=self,
root=root,
source=source,
report=merged,
provenance=provenance,
declarations=declarations,
)
#: :meth:`Schema.compile` exposed as a module-level function.
compile = Schema.compile
@contextmanager
def _policy_compilation(policy: ResourcePolicy | None, *, root_read: bool) -> Iterator[None]:
"""Enter a policy's optional per-compilation closure scope.
:class:`~pyxsd.remote.RemoteResourcePolicy` advertises the hook through the
private ``_pyxsd_compilation_scope`` capability marker, not through the
public ``compilation()`` method's spelling, so a custom policy may define
an unrelated ``compilation`` attribute without breaking compilation. A
policy that wants the same per-compilation state publishes the marker
itself. ``root_read`` is false for an anonymous stream schema, whose
dependencies must still authorize local-to-remote loading.
"""
enter = getattr(policy, "_pyxsd_compilation_scope", None)
if enter is None:
yield
else:
with enter(root_read=root_read):
yield
def _reject_parser_override(
resource_policy: ResourcePolicy | None,
xml_policy: XmlParserPolicy | None,
xml_limits: XmlLimits | None,
) -> None:
"""Refuse parser arguments that a trusted ResourcePolicy would swallow."""
if resource_policy is not None and (xml_policy is not None or xml_limits is not None):
raise ValueError(
"xml_policy and xml_limits cannot be combined with resource_policy: "
"ResourcePolicy.parse() is trusted application code"
)
[docs]
def parse(
xml: str | Path | os.PathLike[str] | IO[str] | ET.Element,
*,
xsd: str | Path | os.PathLike[str] | IO[str] | None = None,
mode: Any = ParseModes.STRICT,
namespace_schemas: dict[str, str | Path] | None = None,
overlay: str | Path | os.PathLike[str] | None = None,
resource_policy: ResourcePolicy | None = None,
xml_policy: XmlParserPolicy | None = None,
xml_limits: XmlLimits | None = None,
) -> "Document":
"""Parses *xml* against its schema and returns a :class:`Document`.
The schema is the one named by ``xsd`` when given; otherwise the
instance's own ``xsi:schemaLocation`` /
``xsi:noNamespaceSchemaLocation`` hints name it, resolved relative
to the instance document's directory (the working directory for
stream sources). An instance that names no schema raises
:class:`~pyxsd.exceptions.PyXSDError`. A hint naming an absolute URI
stays a URI and is never reinterpreted as a local path: a policy
reads it under purpose ``"schema-hint"`` and, without a remote-capable
policy, it raises
:class:`~pyxsd.resource_policy.ResourcePolicyError`.
- ``xml`` - the instance document: a file path, a file object
open for reading, or a parsed ``xml.etree.ElementTree.Element``
used as the document root directly.
- ``xsd`` - the schema to bind against: a path or a file object
open for reading. Defaults to the instance's schema hints.
- ``mode``, ``namespace_schemas``, ``overlay`` - passed to
:meth:`Schema.compile`.
- ``resource_policy`` - a trusted application resource policy for
both the instance and schema documents. Its ``parse()`` owns XML
parsing, so ``xml_policy``/``xml_limits`` cannot be combined with
it.
- ``xml_policy`` / ``xml_limits`` - opt-in parser posture and
structural budgets applied explicitly to the instance and to the
schema compile. They never become an implicit default for later
:meth:`Schema.parse` calls.
"""
_reject_parser_override(resource_policy, xml_policy, xml_limits)
context = NamespaceContext()
# The instance is parsed once, up front: for non-element sources
# its prefix-to-URI bindings land in *context* (shared with the
# schema compile, mirroring the parser's one-context-accumulates-all
# design), and the same tree object is bound below, so the hint
# rewrite a malformed schemaLocation triggers is visible to the
# binding pass.
source_id = source_identifier(xml)
if resource_policy is not None:
try:
data = read_resource(xml, resource_policy, purpose="instance")
tree = resource_policy.parse(data, context)
except (OSError, ET.ParseError) as exc:
raise PyXSDError(f"the xml input could not be read: {exc}") from exc
source_id = data.source_id
elif isinstance(xml, ET.Element):
tree = xml
else:
try:
tree = parse_with_namespaces(xml, context, policy=xml_policy, limits=xml_limits)
except (OSError, ET.ParseError) as e:
raise PyXSDError(f"the xml input could not be read: {e}") from e
# Extra ``xsi:schemaLocation`` pairs beyond the main schema are
# advisory composition hints (strict namespace mode only), read
# from the instance document like the main hint below and resolved
# against the instance document's own directory.
if resource_policy is not None:
assert source_id is not None
base_dir = (
Path(source_id).resolve().parent
if classify_resource(source_id).kind == "local"
else None
)
else:
base_dir = Path(xml).resolve().parent if isinstance(xml, (str, os.PathLike)) else Path.cwd()
pairs = [] if isinstance(xml, ET.Element) else absolute_schema_location_pairs(tree, base_dir)
# A schema the caller named is an application input; a schema selected
# from the instance's own location hint is instance-driven and reads
# under a separate purpose so a policy can authorize it differently.
source_purpose: ResourcePurpose = "schema"
if xsd is None:
_namespace, hint = resolve_schema_hint(tree, base_dir)
if hint is None:
raise PyXSDError(
"no schema file was given and the xml file has no "
"schemaLocation or noNamespaceSchemaLocation tag"
)
xsd = hint
source_purpose = "schema-hint"
schema = Schema.compile(
xsd,
mode=mode,
namespace_schemas=namespace_schemas,
overlay=overlay,
namespace_context=None if isinstance(xml, ET.Element) else context,
schema_location_pairs=pairs,
resource_policy=resource_policy,
xml_policy=xml_policy,
xml_limits=xml_limits,
_source_purpose=source_purpose,
)
if resource_policy is not None:
return schema._bind_tree(tree, xml, source_id=source_id)
# The caller's parser posture and budget were applied to the discovery
# parse above; they are handed to the element bind so the declaration
# re-read honors the same byte budget and effective policy (the
# pre-parsed element itself is never re-parsed).
return schema.parse(tree, xml_policy=xml_policy, xml_limits=xml_limits)
def load_overlay_classes(
schema: Schema,
class_file: str | Path | os.PathLike[str],
fallback_dir: Path | None = None,
) -> None:
"""Loads a file with overlay classes into the schema's class table.
Overlay classes add to and override the schema type classes to
allow for a user to create their own types without changing the
schema file itself. Every loaded class is stamped with the owning
:class:`Schema`.
**Consider this functionality experimental.**
- ``class_file``: a string that specifies the location of a
user-created overlay class file
- ``fallback_dir``: directory searched for ``<class_file>.py`` when
*class_file* does not name an existing file (the historical
module-name fallback against the instance document's directory)
"""
filePath = Path(class_file)
if not filePath.is_file():
# Fall back to the historical behavior of resolving a
# module name against the instance file's directory.
if fallback_dir is not None:
candidate = fallback_dir / f"{class_file}.py"
if candidate.is_file():
filePath = candidate
if not filePath.is_file():
raise ImportError(f"the file '{class_file}' was not found. Please check your spelling.")
module_name = filePath.stem
module = _load_module_from_file(module_name, filePath)
if module is None:
raise ImportError(f"the file '{class_file}' could not be loaded.")
newClasses: dict[str, Any] = {}
for _varName, var in vars(module).items():
if isinstance(var, type) and issubclass(var, SchemaBase) and var is not SchemaBase:
className = getattr(var, "name", None)
if className is None:
warnings.warn(
f"the class {var} must have a 'name' attribute; using '__name__' instead",
PyXSDWarning,
stacklevel=2,
)
className = var.__name__
newClasses[className] = var
logger.debug("Loaded the %s class", className)
schema.classes.update(newClasses)
for cls in newClasses.values():
cls.schema = schema
# Overlay classes can introduce or remove ENTITY/ENTITIES reachability,
# so the lazily cached activation decision is recomputed on next use.
schema._entity_declarations_needed = None
def _compile_into_context(
ctx: CompositionContext,
xsd_file: str | Path | os.PathLike[str] | IO[str] | None,
host: Any,
) -> Schema:
"""Runs the schema-compilation pipeline into *ctx*.
The single shared body behind :meth:`Schema.compile`. *host* is
the parser-like object the
ElementRepresentative APIs read (``report``, ``mode``,
``components``, ``classes``, ``namespaceContext``); the container
fields of *ctx* alias the host's own containers, so the compiled
products stay visible to both. Returns the built :class:`Schema`
with every generated class stamped with it.
"""
logger.debug("Sending the schema file to the ElementTree Parser...")
# Everything reported until the instance document is parsed is a
# schema-compilation problem: composition, ER building, and class
# generation.
ctx.report.phase = "schema"
# The report rides the active context for the whole run, so
# diagnostics raised anywhere in the pipeline (ER building, class
# building, the declaration sweeps) resolve it without needing the
# parser passed down; the previous value is restored afterwards.
with active_context(ctx.schema_context), context_report(ctx.schema_context, ctx.report):
if ctx.resource_policy is not None:
assert ctx.source_data is not None
try:
root = ctx.resource_policy.parse(ctx.source_data, ctx.namespace_context)
except ET.ParseError as exc:
raise PyXSDError(f"the schema file is not well-formed XML: {exc}") from exc
elif isinstance(xsd_file, (str, os.PathLike)):
try:
with open(xsd_file, "rb") as schemaFile:
root = parse_with_namespaces(
schemaFile,
ctx.namespace_context,
policy=ctx.xml_policy,
limits=ctx.xml_limits,
)
except OSError as e:
raise PyXSDError(f"the schema file could not be opened: {e}") from e
except ET.ParseError as e:
raise PyXSDError(f"the schema file is not well-formed XML: {e}") from e
else:
# The schema must be a file object here: a ``None`` schema
# without a location hint is rejected by the caller.
assert xsd_file is not None
try:
root = parse_with_namespaces(
xsd_file,
ctx.namespace_context,
policy=ctx.xml_policy,
limits=ctx.xml_limits,
)
except ET.ParseError as e:
raise PyXSDError(f"the schema file is not well-formed XML: {e}") from e
logger.debug("Sending the schema ElementTree to the ElementRepresentative module...")
# XSD 1.1 §4.2.2 conditional inclusion runs on every schema document
# before anything else, so the element-representative walk never sees
# a declaration a ``vc:*`` selector excludes. Included and imported
# documents are filtered as they are parsed (``parse_included_schema``).
record_declarations(root, source_identifier(xsd_file), ctx.declaration_origins)
apply_conditional_inclusion(root, ctx.namespace_context, ctx.report, ctx.processor_version)
check_namespace_attribute_values(ctx, root)
if ctx.processor_version == decimal.Decimal("1.0"):
check_xsd10_vocabulary(root, ctx.report)
base, visited = schema_composition_context(ctx, xsd_file)
# Documents already fully composed; their components must not be
# spliced twice (diamond includes) and a repeat encounter is not
# an error.
ctx.composed_documents = set(visited)
# A namespace is satisfied if a schema for it was supplied up
# front, so such a namespace-only import is not unresolved.
ctx.resolved_imports.update(ns for ns in ctx.namespace_schemas if ns)
ctx.resolved_imports.update(
item.namespace for item in ctx.additional_schemas if item.namespace
)
mainTargetNamespace = root.get("targetNamespace")
if mainTargetNamespace:
ctx.composed_target_namespaces.add(mainTargetNamespace)
splice_composed_schemas(ctx, root, base, visited, mainDocument=True)
splice_additional_schemas(ctx, root, base, visited)
# Post-composition audit (XSD 1.1 §4.2.6): every QName
# component reference is judged by its own document's authored
# imports, before any injected built-in declarations join the
# tree and before the ER run snapshots it.
audit_direct_imports(ctx, root)
if getattr(ctx.mode, "namespaces", "legacy") == "strict":
inject_xml_namespace_attributes(ctx, root)
inject_xsi_namespace_attributes(ctx, root)
inject_xlink_namespace_attributes(ctx, root)
elif ctx.xml_namespace_imported and XML_NS not in ctx.resolved_imports:
# A namespace-only ``xs:import`` of the XML namespace (the
# W3C schema-for-schemas' own spelling, XSD 1.1 §4.2.3):
# no document was composed for it, so its built-in
# attribute declarations are what references resolve
# against. When a document *was* composed, its own
# declarations are used instead and nothing is injected.
inject_xml_namespace_attributes(ctx, root)
# XSD 1.1 allows a local element or attribute declaration to state
# its own target namespace, but only inside an xs:restriction
# (§3.3.2.1, §3.2.2). Register the value as a namespace override
# before the ER run snapshots ``namespaceOverrides`` so the
# declaration's expanded name reflects it.
apply_local_target_namespaces(ctx, root)
check_form_defaults(ctx, root)
# Stage this compile's snapshot on its context before the ER
# run so imported declarations report their own target namespace
# and form defaults.
ctx.schema_context.namespace_overrides = dict(ctx.namespace_overrides)
ctx.schema_context.component_namespaces = dict(ctx.component_namespaces)
ctx.schema_context.injected_builtin_ids = set(ctx.injected_builtin_ids)
ctx.schema_context.form_defaults = dict(ctx.form_defaults)
ctx.schema_context.xpath_default_namespaces = dict(ctx.xpath_default_namespaces)
schemaER = ElementRepresentative.factory(root, None)
if schemaER is None or schemaER.__class__.__name__ != "Schema":
# A document that is not an XML Schema at all (for example
# an instance document passed as the schema) must surface as
# an input error, not as an AttributeError on ``None``.
raise PyXSDError(f"schema document root element is not xs:schema: {root.tag!r}")
# Attach the host and the compilation report to the schema ER:
# the host so class building and schema-time QName resolution
# can reach the parser-like containers, the report so
# schema-reference problems recorded during the ER walk and the
# declaration checks land on the validation report.
schemaER.host = host
schemaER.report = ctx.report
# The captured prefix bindings let every declaration resolve the
# QNames written in its own document. Install them before the
# declaration checks run: the atomicity check resolves
# ``itemType``/``memberTypes`` through the in-scope namespaces.
schemaER.namespaceContext = ctx.namespace_context
# Expand the 1.1 notQName names on every wildcard before the
# declaration walk: the attribute-wildcard derivation check runs
# when a complex type is visited, ahead of the AnyAttribute ERs
# that declared the specs, and must see the expanded names.
refine_wildcard_specs(ctx, schemaER)
# This compile owns the component table the ER run registered
# into; expose it on the host and on the context so registry
# lookups (xsi:type dispatch, tests, and the declaration-issue
# walk's type resolution) use this compile's declarations rather
# than a previous one's.
host.components = schemaER.components
ctx.schema_context.components = host.components
# Module-level lookups from here on (ElementRepresentative
# .getFromName, the registry proxy) see this compile's table.
remember_components(host.components)
report_declaration_issues(ctx, schemaER, host)
check_keyref_references(ctx, schemaER)
# After the declaration walk (which parses every explicit
# ``xs:openContent`` and the host document's
# ``xs:defaultOpenContent``), attach the applicable schema default
# to each complex type that declares none of its own.
apply_default_open_content(ctx, schemaER)
# XSD 1.1 §3.1.2: attach each schema document's default attribute
# group to the complex types it declares, unless the type sets
# ``defaultAttributesApply="false"``. Runs before class building so
# ``resolveAttributeGroupRefs`` folds the group in with the
# explicit references.
apply_default_attributes(ctx, schemaER)
# With every type's effective open content settled (explicit or
# inherited default), check the open-content derivation rules
# (mode rank and wildcard subset for restriction/extension).
report_open_content_derivations(ctx, schemaER, host)
# The schema root is itself the instance class used to dispatch
# the document root's element declarations.
ctx.classes["schema"] = schemaER.clsFor(host)
# Build a generated class for every named type in the
# compile-owned component table. The per-schema ``simpleTypes`` /
# ``complexTypes`` dicts are keyed by local name, so composed
# schemas that reuse a local name in different namespaces (very
# common in OOXML) collide there and lose declarations. The
# component table keeps one entry per expanded name.
built: set[int] = set()
for entries in host.components.values():
for typeER in entries:
if componentKind(typeER) != "type" or id(typeER) in built:
continue
if getattr(typeER, "_alternativeInline", False):
# An ``xs:alternative``'s inline type is built on
# demand (when a test selects it, or when the
# derivation check needs it), never eagerly: building
# every one here would surface an unimplemented base
# as a schema error even for alternatives the instance
# phase never selects.
continue
built.add(id(typeER))
cls = typeER.clsFor(host)
ctx.classes[typeER.name] = cls
if typeER.expandedName:
ctx.classes[typeER.expandedName] = cls
logger.debug("Class created for the %s type...", typeER.name)
build_substitution_groups(ctx, schemaER, host)
check_substitution_group_exclusions(ctx, schemaER, host)
check_value_constraints(ctx, schemaER)
check_type_references(ctx, schemaER)
check_notation_uses(ctx, schemaER, host)
check_simple_content_restriction_base(ctx, schemaER, host)
check_content_kind_derivation(ctx, schemaER, host)
check_alternatives(ctx, schemaER)
check_alternative_table_edc(ctx, schemaER)
check_conditional_type_substitutable(ctx, schemaER, host)
check_group_redefine_restrictions(ctx, schemaER, host)
schema = Schema(
classes=ctx.classes,
components=host.components,
namespace_context=ctx.namespace_context,
mode=ctx.mode,
target_namespace=mainTargetNamespace,
report=ctx.report,
source=xsd_file,
schema_context=ctx.schema_context,
)
# The Schema object only exists after the harvest, so the
# back-reference is stamped in one pass here (overlay classes
# are stamped by ``load_overlay_classes``).
for cls in ctx.classes.values():
cls.schema = schema
schema._host = host
schema.resource_policy = ctx.resource_policy
schema.declaration_origins = ctx.declaration_origins
schema.xsd_version = str(host.processor_version)
# Classes built lazily after the compile (an ``xs:alternative``'s
# inline type selected at binding time) stamp their owner from
# the schema ER, which outlives the compile on the component
# table.
schemaER.compiledSchema = schema
return schema