Skip to content

Model

model

Package containing models, the pythonic representation of feature classes.

They inherit from FeatureType or DataType depending on their UML stereotype. Only a FeatureType carries an id of its own, and only it is written to GML, the coretable or JSON-FG; a DataType is a value object nested inside a feature, reaching a document only inline inside its owner. Both extend BaseType -- the Pydantic BaseModel plus the class metadata neither kind can do without -- which is an internal base: annotate against FeatureType, DataType, or FeatureType | DataType where either is valid. Classes are resolved via an Appschema instance. A feature collection is represented by the FeatureCollection class.

Example

Load the BP_Plan model for XPlanung v6.0 and instantiate it with some data:

from xmas_core.model import Appschema

plan = Appschema.from_prefix("xplan", "6.0").model_factory("BP_Plan")
instance = plan.model_validate(
    {
        "name": "Testplan",
        "gemeinde": [
            {
                "ags": "1234"
            }
        ],
        "raeumlicherGeltungsbereich": {
            "srid": 25832,
            "wkt": <WKT-String>
        }
    }
)

DROP_INVALID_REFS = 'drop_invalid_refs' module-attribute

Validation context key asking FeatureCollection to drop invalid references.

INVALID_REFS = 'invalid_refs' module-attribute

Validation context key the dropped references are collected under.

Appschema

Bases: BaseModel

Models a supported appschema.

Used to access metadata like name, version and description for an appschema.

During instantion, validation if an appschema is supported occurs, based on the modules in appschema subdirectory.

The instance stores a reference to the appschema's module, which is used in the model_factory method the retrieve appschema classes.

Usage example
# get a featuretype from appschema
appschema = Appschema.from_prefix(prefix="xplan", version="6.0")
plan = appschema.model_factory(name="BP_Plan")
# show list of supported appschemas
Appschema.supported_appschemas()
# get an enum of supported appschemas
SupportedAppschemas = Appschema.enum()
SupportedAppschemas.XPLAN_6_0.value # -> XPlanGML 6_0

code property

The enum code of the appschema.

max_containment_depth cached property

Longest whole -> part chain from a top-level owner, i.e. how deep a plan can nest.

An upper bound on the hop distance between a plan and anything belonging to it, which is what a bounded graph traversal needs in order not to truncate.

It is the longest path, not the shortest: shortest-path distance conflates a type with its instances, and would put XP_TextAbschnitt one hop away because BP_Plan.texte names that type, while an abschnitt referenced only by an object of the plan actually sits three hops out.

The bound is therefore safe but not tight -- most links on the longest chain also hang directly off a Bereich, so no instance need sit that deep -- and it says nothing about features reached through a part shared with another plan, which no depth excludes.

Returns:

Type Description
int

The longest chain length, or 0 for an appschema whose types own nothing.

Raises:

Type Description
RecursionError

A type transitively owns itself, so no finite depth bounds the traversal. Every supported appschema is acyclic.

navigable_roles cached property

Every association role of this appschema that becomes a refs edge.

One row per (source feature type, role, concrete target type). This is the Python side of the navigable_roles_config table the coretable traversal functions read, and scripts/navigable_roles.py renders that table's seed from it -- so the seed is generated rather than a second hand-maintained encoding of the same graph.

Only a FeatureType is stored as its own coretable row and so produces refs edges. Data types (XP_VerbundenerPlan, LP_VorschlagIntegration*, ...) are serialized into the parent's properties JSON, so their roles never become refs rows.

Association typenames are pre-expanded to concrete classes by the generator, so nothing here walks subclasses.

Returns:

Type Description
frozenset[NavigableRole]

The edges as NavigableRole tuples. Unordered: the walk runs over a frozenset of

frozenset[NavigableRole]

role names, so a caller that renders them sorts first.

ownership_edges cached property

The whole -> part containment graph of this appschema, keyed by owner.

One entry per feature type that existentially owns at least one other (an association whose dependent_part is set), mapping it to the types it owns. Association typenames resolve to concrete classes throughout the supported appschemas, so this is the concrete graph, not one that stops at an abstract supertype.

The ownership rows of navigable_roles keyed by owner -- one walk serves both. Ownership is between stored objects, so restricting the walk to FeatureType drops no edge.

Returns:

Type Description
dict[str, frozenset[str]]

Owner feature type name -> the feature type names it existentially owns.

top_level_featuretypes cached property

Names of the top-level owner feature types of this appschema.

A top-level owner existentially owns at least one part (an association whose dependent_part is set) but is itself never the part of another owner: the containment/deletion roots, e.g. BP_Plan. Derived from the appschema's association metadata as the owner-set minus the part-set.

Returns:

Type Description
list[str]

The owner feature type names, sorted.

version property

The version of the appschema in <major>.<minor> format.

enum() cached classmethod

Return an enumeration of supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def enum(cls) -> type[_SupportedAppschema]:
    """Return an enumeration of supported appschemas."""
    members = {
        f"{appschema.prefix.upper()}_{appschema.full_version.major}_{appschema.full_version.minor}": (
            f"{appschema.prefix.upper()}_{appschema.version.replace('.', '_')}",
            appschema,
        )
        for appschema in cls.supported_appschemas()
    }
    # The functional API, called on the metaclass: a checker reads a call on an enum class
    # that defines `__new__` as constructing one member, and types the result `type[Enum]`.
    return cast(
        type[_SupportedAppschema],
        EnumType.__call__(_SupportedAppschema, "SupportedAppschemas", members),
    )

from_enum(code) classmethod

Return an Appschema instance from enum code.

Parameters:

Name Type Description Default
code str

an appschema code, e.g. XPLAN_6_0

required
Source code in xmas_core/model/meta.py
@classmethod
def from_enum(cls, code: str) -> Appschema:
    """Return an Appschema instance from enum code.

    Args:
        code: an appschema code, e.g. `XPLAN_6_0`
    """
    try:
        return cls.enum()[code].appschema
    except KeyError:
        raise NotImplementedError(f"no appschema with code {code!r}") from None

from_module(module_name) cached classmethod

Builds an Appschema instance from module name.

Parameters:

Name Type Description Default
module_name str

the fully qualified module name

required
Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def from_module(cls, module_name: str) -> Appschema:
    """Builds an Appschema instance from module name.

    Args:
        module_name: the fully qualified module name
    """
    try:
        # module_name is APPSCHEMA_MODULE_NAMES-derived or a cls.__module__, never input
        # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
        module = import_module(module_name)
        model_cls: type[RootModel[Any]] = module.Model
    except (ImportError, AttributeError) as exc:
        raise RuntimeError(
            f"Unable to resolve appschema metadata for module {module_name!r}"
        ) from exc

    if not issubclass(model_cls, RootModel):
        raise TypeError(f"expected RootModel, got {model_cls!r}")

    metadata = model_cls.model_fields["root"]

    return cls.model_validate(
        _schema_extra(metadata)
        | {"description": metadata.description, "module": module}
    )

from_namespace(namespace) classmethod

Builds an Appschema instance from the appschema's namespace.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
namespace str

the namespace URI of the appschema

required
Source code in xmas_core/model/meta.py
@classmethod
def from_namespace(cls, namespace: str) -> Appschema:
    """Builds an Appschema instance from the appschema's namespace.

    Might return a compatible appschema version if no exact match is found.

    Args:
        namespace: the namespace URI of the appschema
    """
    uri = AnyUrl(namespace)
    candidates: list[Appschema] = []
    for appschema in cls.supported_appschemas():
        if appschema.namespace_uri == uri:
            return appschema
        elif str(appschema.namespace_uri)[:-1] == namespace[:-1]:
            candidates.append(appschema)
    if candidates:
        return max(candidates)
    raise NotImplementedError(f"no appschema with namespace {namespace!r}")

from_prefix(prefix, version) classmethod

Builds an Appschema instance from the appschema's prefix and version.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
prefix str

the namespace prefix of the appschema, e.g. xplan or xtrasse

required
version str

the version of the appschema; major and minor are required

required
Source code in xmas_core/model/meta.py
@classmethod
def from_prefix(cls, prefix: str, version: str) -> Appschema:
    """Builds an Appschema instance from the appschema's prefix and version.

    Might return a compatible appschema version if no exact match is found.

    Args:
        prefix: the namespace prefix of the appschema, e.g. `xplan` or `xtrasse`
        version: the version of the appschema; major and minor are required
    """
    sem_version = Version.parse(version, optional_minor_and_patch=True)
    versions = []
    candidates = []
    for appschema in filter(
        lambda appschema: appschema.prefix == prefix, cls.supported_appschemas()
    ):
        # if exact match, return
        if (
            sem_version.major == appschema.full_version.major
            and sem_version.minor == appschema.full_version.minor
        ):
            return appschema
        # if versions are compatible (modules minor version higher or equal), add to candidates
        elif sem_version.is_compatible(appschema.full_version):
            candidates.append(appschema)
        else:
            versions.append(
                f"{appschema.full_version.major}.{appschema.full_version.minor}"
            )
    # return appschema with newer minor version, if found
    # TODO: ensure minor version compatibility without version migration, e.g. via model_validator
    # - in GML it's automatic though
    if candidates:
        return max(candidates)
    if versions:
        e = NotImplementedError(
            f"version {version!r} not supported for prefix {prefix!r}"
        )
        e.add_note(f"available versions: {', '.join(sorted(versions))}")
        raise e
    else:
        raise NotImplementedError(f"no appschema with prefix {prefix!r}")

model_factory(name, expect=None)

model_factory(name: str) -> type[FeatureType | DataType]
model_factory(name: str, expect: type[T]) -> type[T]

Factory method for retrieving the corresponding pydantic model representation of a feature class.

Parameters:

Name Type Description Default
name str

name of the feature class or enumeration

required
expect type[BaseType] | None

the base class the result has to derive from. Pass FeatureType where only an identifiable feature will do, so the result is typed as one and a data type raises instead of failing later on a missing id. Defaults to BaseType, which accepts either kind - and since every appschema class is exactly one of the two (pinned by test_every_appschema_class_is_a_featuretype_or_a_datatype), the bare call is typed as type[FeatureType | DataType] rather than the abstract base, so its result can be passed on where either kind is expected.

None

Raises:

Type Description
ValueError

requested class not found, or it is not a subclass of expect.

Returns:

Type Description
type[BaseType]

The concrete class, typed as the requested expect.

Source code in xmas_core/model/meta.py
def model_factory(
    self, name: str, expect: type[BaseType] | None = None
) -> type[BaseType]:
    """Factory method for retrieving the corresponding pydantic model representation of a feature class.

    Args:
        name: name of the feature class or enumeration
        expect: the base class the result has to derive from. Pass `FeatureType` where only
            an identifiable feature will do, so the result is typed as one and a data type
            raises instead of failing later on a missing `id`. Defaults to `BaseType`,
            which accepts either kind - and since every appschema class is exactly one of
            the two (pinned by `test_every_appschema_class_is_a_featuretype_or_a_datatype`),
            the bare call is typed as `type[FeatureType | DataType]` rather than the
            abstract base, so its result can be passed on where either kind is expected.

    Raises:
        ValueError: requested class not found, or it is not a subclass of `expect`.

    Returns:
        The concrete class, typed as the requested `expect`.
    """
    try:
        cls = getattr(self.module, name)
    except AttributeError:
        raise ValueError(f"featuretype {name!r} not found for appschema {self!r}")
    return _ensure_subclass(cls, expect if expect is not None else BaseType)

supported_appschemas() cached classmethod

Return a list of currently supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def supported_appschemas(cls) -> list[Appschema]:
    """Return a list of currently supported appschemas."""
    return [
        cls.from_module(f"{__package__}.appschema.{module_name}")
        for module_name in APPSCHEMA_MODULE_NAMES
    ]

DataType

Bases: BaseType

A value object nested inside a feature, with no identity of its own.

It reaches a document only as the value of some feature's property, is written and read inline by its owner's encoder and lands in the coretable inside that feature's properties JSON, so it has no id, is never a collection member, and never stands for a GML element, a coretable row or a JSON-FG object of its own. referenzURL / georefURL are declared on data types only, so their validator and serializer live here.

FeatureCollection

Bases: BaseModel

Container for features that provides validation of references.

The features are stored in a dictionary with their ID as key and the feature instance as value.

plans property

The collection's plans - its top-level features - by id.

Derived from features on every access rather than stored, so it cannot be passed in and cannot name a plan the collection does not hold, however the features change.

from_features(features, srid, appschema, *, context=None) classmethod

Builds a collection, passing context to the reference check.

The one constructor every repository uses, so the read options a caller can set - DROP_INVALID_REFS above all - reach the validator the same way whichever format was read.

Parameters:

Name Type Description Default
features dict[UUID, FeatureType]

The features, keyed by id.

required
srid int | None

The collection's spatial reference system identifier. None, for a document whose SRS could not be resolved, fails validation.

required
appschema Appschema

The appschema every feature belongs to.

required
context dict[str, Any] | None

Pydantic validation context; see check_references_and_srs.

None

Returns:

Type Description
FeatureCollection

The validated collection.

Source code in xmas_core/model/collection.py
@classmethod
def from_features(
    cls,
    features: dict[UUID, FeatureType],
    srid: int | None,
    appschema: Appschema,
    *,
    context: dict[str, Any] | None = None,
) -> FeatureCollection:
    """Builds a collection, passing `context` to the reference check.

    The one constructor every repository uses, so the read options a caller can set -
    `DROP_INVALID_REFS` above all - reach the validator the same way whichever format
    was read.

    Args:
        features: The features, keyed by id.
        srid: The collection's spatial reference system identifier. `None`, for a
            document whose SRS could not be resolved, fails validation.
        appschema: The appschema every feature belongs to.
        context: Pydantic validation context; see `check_references_and_srs`.

    Returns:
        The validated collection.
    """
    return cls.model_validate(
        {"features": features, "srid": srid, "appschema": appschema},
        context=context,
    )

get_features()

Yields features stored in the collection.

Source code in xmas_core/model/collection.py
def get_features(self) -> Iterator[FeatureType]:
    """Yields features stored in the collection."""
    return (feature for feature in self.features.values())

get_single_plans()

Yields FeatureCollection objects for every plan in the original collection.

Each one holds exactly one plan, which its plans names: a top-level feature type is never the part of another, so no closure reaches a second.

A plan collection is the containment closure of a top-level feature type, derived from the appschema's dependent_part metadata rather than from hardcoded role names: a role marked target points at a part the feature owns, a role marked source points back at its owner. Both directions are followed, since a file may populate only one of them.

A plan is self-contained, so a reference leaving its collection is invalid data and raises when the yielded collection validates its references.

Raises:

Type Description
ValueError

If the appschema declares no dependent_part metadata, so no containment closure can be derived. Every supported appschema declares it.

Source code in xmas_core/model/collection.py
def get_single_plans(self) -> Iterator[FeatureCollection]:
    """Yields FeatureCollection objects for every plan in the original collection.

    Each one holds exactly one plan, which its `plans` names: a top-level feature type is
    never the part of another, so no closure reaches a second.

    A plan collection is the containment closure of a top-level feature type, derived
    from the appschema's `dependent_part` metadata rather than from hardcoded role
    names: a role marked `target` points at a part the feature owns, a role marked
    `source` points back at its owner. Both directions are followed, since a file may
    populate only one of them.

    A plan is self-contained, so a reference leaving its collection is invalid data and
    raises when the yielded collection validates its references.

    Raises:
        ValueError: If the appschema declares no `dependent_part` metadata, so no
            containment closure can be derived. Every supported appschema declares it.
    """
    if not self.appschema.top_level_featuretypes:
        raise ValueError(
            f"appschema {self.appschema.full_name} declares no ownership metadata, "
            "single plans cannot be derived"
        )
    return self._single_plans()

make_copy(with_id_map=False)

Return a copy of the collection with new IDs.

All feature IDs are renewed and respective references are updated.

Parameters:

Name Type Description Default
with_id_map bool

whether to additionally return a map of old IDs to new IDs

False
Source code in xmas_core/model/collection.py
def make_copy(
    self, with_id_map: bool = False
) -> FeatureCollection | tuple[FeatureCollection, dict[UUID, UUID]]:
    """Return a copy of the collection with new IDs.

    All feature IDs are renewed and respective references are updated.

    Args:
        with_id_map: whether to additionally return a map of old IDs to new IDs
    """
    id_map = {key: uuid7() for key in self.features}
    new_features = {}
    for old_feature in self.features.values():
        new_id = id_map[old_feature.id]
        new_feature = old_feature.model_copy(deep=True)
        new_feature.id = new_id
        for assocation in old_feature.get_associations():
            old_value = getattr(old_feature, assocation)
            if isinstance(old_value, UUID):
                new_value = id_map[old_value]
                setattr(new_feature, assocation, new_value)
            elif isinstance(old_value, list):
                new_list = [
                    id_map[item] if isinstance(item, UUID) else item
                    for item in old_value
                ]
                setattr(new_feature, assocation, new_list)
        new_features[new_id] = new_feature

    new_collection = FeatureCollection(
        features=new_features,
        srid=self.srid,
        appschema=self.appschema,
    )
    if with_id_map:
        return new_collection, id_map
    else:
        return new_collection

FeatureType

Bases: BaseType

An independently identifiable object: it carries a UUID of its own.

Everything stored as a row in the coretable, held as a member of a FeatureCollection or returned by a repository is one of these. id is declared here so it resolves on a value typed as this class; the generated appschema roots re-declare it with the same annotation and their own json_schema_extra, which is what get_property_info("id") reads.

It also carries everything a data type has no use for: the geometry validator and the hatGenerAttribut machinery - verified over every appschema module, a Geometry field and hatGenerAttribut occur on feature types only.

It is also the only kind that defers its schema build. Pydantic inlines a nested model's schema into its parent, so a class reached only as a field of another one is validated and constructed without a serializer of its own ever being built - it keeps a MockValSer, and anything that serializes such an instance through pydantic_core raises 'MockValSer' object is not an instance of 'SchemaSerializer'. A feature type is never nested: it is always validated directly, through model_factory then model_validate, so deferring it is free. A data type and the value types in appschema.definitions are never anything but nested, so deferring those only cost them their serializer. It is also where the time is - per appschema module, building every feature type costs 513 ms against 19 ms for every data type and 2 ms for all of definitions.

get_geom_field() cached classmethod

Returns the classes geometry field name, if any.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_field(cls) -> str | None:
    """Returns the classes geometry field name, if any."""
    for name in cls.model_fields:
        if cls.get_property_info(name).stereotype == "Geometry":
            return name
    return None

get_geom_srid()

Returns the object's geometry's SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_srid(self) -> int | None:
    """Returns the object's geometry's SRID, if any."""
    return geom.srid if (geom := self._geom()) else None

get_geom_types() cached classmethod

Returns the types of the geometry attribute.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_types(cls) -> frozenset[type[GeometryType]]:
    """Returns the types of the geometry attribute."""
    if geom_field := cls.get_geom_field():
        geom_annotation = cls.model_fields[geom_field].annotation
        if not (args := get_args(geom_annotation)):
            return frozenset({cast(type[GeometryType], geom_annotation)})
        return frozenset(arg for arg in args if arg is not NoneType)
    return frozenset()

get_geom_wkt()

Returns the object's eWKT geometry's WKT representation withouth SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_wkt(self) -> str | None:
    """Returns the object's eWKT geometry's WKT representation withouth SRID, if any."""
    return geom.wkt if (geom := self._geom()) else None

InvalidReference

Bases: BaseModel

An association reference a collection cannot support.

Either the referenced UUID names no feature in the collection, or the reference is an external URL. Associations are intra-document by design, so neither resolves to a feature and both are unsupported.

base

Contains BaseType and its two kinds, FeatureType and DataType.

The classes provide some utility, (de-)serialization and/or validation methods to simplify access and manipulation. What resolves them from a name is Appschema in meta, and what holds many of them is FeatureCollection in collection, so that this module stays about a single hierarchy.

POLYMORPHIC_DISCRIMINATOR = 'datatype' module-attribute

Serialization and validation context key for a polymorphic discriminator.

AssocInfo

Bases: BaseModel

Association metadata returned inside PropertyInfo.

BaseEnum

Bases: StrEnum

Base class for appschema enumerations.

Extends StrEnum with additional metadata like description and alias.

BaseType

Bases: _ConfiguredBaseModel

Base class for application schema classes.

It extends pydantic BaseModel with the class metadata every appschema class needs - get_name, appschema, get_property_info, get_associations, the get_geom_* family. The encodings in codec hold whole features, and a data type reaches GML only inline inside its owner, which the GML codec encodes from this metadata alone.

Concrete appschema classes never derive from this class directly: they derive from FeatureType or DataType, which is what makes id resolvable on the former. This class stays the common supertype for the places where either kind is valid, and is named nowhere outside this module - FeatureType | DataType is how callers spell it.

appschema() cached classmethod

Return metadata about the application schema for the feature class.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def appschema(cls) -> Appschema:
    """Return metadata about the application schema for the feature class."""
    # circular: `meta` imports this module. Cached, so the import runs once per class.
    from xmas_core.model.meta import Appschema

    return Appschema.from_module(cls.__module__)

get_associations() cached classmethod

Returns the classes association fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_associations(cls) -> frozenset[str]:
    """Returns the classes association fields."""
    return frozenset(
        assoc
        for assoc in cls.model_fields
        if cls.get_property_info(assoc).stereotype == "Association"
    )

get_measure_fields() cached classmethod

Returns the classes measure fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_measure_fields(cls) -> frozenset[str]:
    """Returns the classes measure fields."""
    return frozenset(
        measure
        for measure in cls.model_fields
        if cls.get_property_info(measure).stereotype == "Measure"
    )

get_name() cached classmethod

Returns the canonical name of the class.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_name(cls) -> str:
    """Returns the canonical name of the class."""
    return cls.__name__

get_polymorphic_fields() cached classmethod

The data type fields that admit more than one class.

A mapping does not say which of them it was built from, and where the classes share their fields the reader cannot tell either, so these are the fields whose values carry their class as datatype when a dump is asked for it.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_polymorphic_fields(cls) -> frozenset[str]:
    """The data type fields that admit more than one class.

    A mapping does not say which of them it was built from, and where the classes
    share their fields the reader cannot tell either, so these are the fields whose
    values carry their class as `datatype` when a dump is asked for it.
    """
    return frozenset(
        name
        for name in cls.model_fields
        if (prop_info := cls.get_property_info(name)).stereotype == "DataType"
        and isinstance(prop_info.typename, list)
    )

get_property_info(name) cached classmethod

Property information.

Parameters:

Name Type Description Default
name str

The property's name.

required

Returns:

Type Description
PropertyInfo

A typed dataclass holding the information.

Raises:

Type Description
AttributeError

The name was not found in the model fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_property_info(cls, name: str) -> PropertyInfo:
    """Property information.

    Args:
        name: The property's name.

    Returns:
        A typed dataclass holding the information.

    Raises:
        AttributeError: The name was not found in the model fields.
    """
    try:
        extra_info = _schema_extra(cls.model_fields[name])
    except KeyError:
        raise AttributeError(f"Unknown property: {name}")
    else:
        stereotype: Final = extra_info["stereotype"]
        typename: str | list[str] = extra_info["typename"]
        return PropertyInfo(
            stereotype=stereotype,
            typename=typename,
            list=not extra_info["multiplicity"].endswith("1"),
            nullable=extra_info["multiplicity"].startswith("0"),
            uom=extra_info.get("uom", None),
            enum=_ensure_subclass(
                getattr(cls.appschema().module, str(typename)), BaseEnum
            )
            if stereotype == "Enumeration"
            else None,
            assoc_info=None
            if stereotype != "Association"
            else AssocInfo(
                reverse=extra_info.get("reverseProperty"),
                source_or_target=extra_info.get("sourceOrTarget"),
                dependent_part=extra_info.get("dependent_part"),
            ),
        )

DataType

Bases: BaseType

A value object nested inside a feature, with no identity of its own.

It reaches a document only as the value of some feature's property, is written and read inline by its owner's encoder and lands in the coretable inside that feature's properties JSON, so it has no id, is never a collection member, and never stands for a GML element, a coretable row or a JSON-FG object of its own. referenzURL / georefURL are declared on data types only, so their validator and serializer live here.

FeatureType

Bases: BaseType

An independently identifiable object: it carries a UUID of its own.

Everything stored as a row in the coretable, held as a member of a FeatureCollection or returned by a repository is one of these. id is declared here so it resolves on a value typed as this class; the generated appschema roots re-declare it with the same annotation and their own json_schema_extra, which is what get_property_info("id") reads.

It also carries everything a data type has no use for: the geometry validator and the hatGenerAttribut machinery - verified over every appschema module, a Geometry field and hatGenerAttribut occur on feature types only.

It is also the only kind that defers its schema build. Pydantic inlines a nested model's schema into its parent, so a class reached only as a field of another one is validated and constructed without a serializer of its own ever being built - it keeps a MockValSer, and anything that serializes such an instance through pydantic_core raises 'MockValSer' object is not an instance of 'SchemaSerializer'. A feature type is never nested: it is always validated directly, through model_factory then model_validate, so deferring it is free. A data type and the value types in appschema.definitions are never anything but nested, so deferring those only cost them their serializer. It is also where the time is - per appschema module, building every feature type costs 513 ms against 19 ms for every data type and 2 ms for all of definitions.

get_geom_field() cached classmethod

Returns the classes geometry field name, if any.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_field(cls) -> str | None:
    """Returns the classes geometry field name, if any."""
    for name in cls.model_fields:
        if cls.get_property_info(name).stereotype == "Geometry":
            return name
    return None

get_geom_srid()

Returns the object's geometry's SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_srid(self) -> int | None:
    """Returns the object's geometry's SRID, if any."""
    return geom.srid if (geom := self._geom()) else None

get_geom_types() cached classmethod

Returns the types of the geometry attribute.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_types(cls) -> frozenset[type[GeometryType]]:
    """Returns the types of the geometry attribute."""
    if geom_field := cls.get_geom_field():
        geom_annotation = cls.model_fields[geom_field].annotation
        if not (args := get_args(geom_annotation)):
            return frozenset({cast(type[GeometryType], geom_annotation)})
        return frozenset(arg for arg in args if arg is not NoneType)
    return frozenset()

get_geom_wkt()

Returns the object's eWKT geometry's WKT representation withouth SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_wkt(self) -> str | None:
    """Returns the object's eWKT geometry's WKT representation withouth SRID, if any."""
    return geom.wkt if (geom := self._geom()) else None

GeometryType

Bases: Protocol

A feature geometry type.

Attributes:

Name Type Description
srid int

The spatial reference system identifier.

wkt str

A WKT string.

PropertyInfo

Bases: BaseModel

UML model information of the given property.

type_set cached property

Valid value types for the property as a set.

is_type_ok(typename)

Return whether a single typename or all typenames in a list are valid.

Source code in xmas_core/model/base.py
def is_type_ok(self, typename: str | builtins.list[str]) -> bool:
    """Return whether a single typename or all typenames in a list are valid."""
    test_types = set(typename) if isinstance(typename, list) else {typename}
    return test_types.issubset(self.type_set)

BaseType

Bases: _ConfiguredBaseModel

Base class for application schema classes.

It extends pydantic BaseModel with the class metadata every appschema class needs - get_name, appschema, get_property_info, get_associations, the get_geom_* family. The encodings in codec hold whole features, and a data type reaches GML only inline inside its owner, which the GML codec encodes from this metadata alone.

Concrete appschema classes never derive from this class directly: they derive from FeatureType or DataType, which is what makes id resolvable on the former. This class stays the common supertype for the places where either kind is valid, and is named nowhere outside this module - FeatureType | DataType is how callers spell it.

appschema() cached classmethod

Return metadata about the application schema for the feature class.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def appschema(cls) -> Appschema:
    """Return metadata about the application schema for the feature class."""
    # circular: `meta` imports this module. Cached, so the import runs once per class.
    from xmas_core.model.meta import Appschema

    return Appschema.from_module(cls.__module__)

get_associations() cached classmethod

Returns the classes association fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_associations(cls) -> frozenset[str]:
    """Returns the classes association fields."""
    return frozenset(
        assoc
        for assoc in cls.model_fields
        if cls.get_property_info(assoc).stereotype == "Association"
    )

get_measure_fields() cached classmethod

Returns the classes measure fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_measure_fields(cls) -> frozenset[str]:
    """Returns the classes measure fields."""
    return frozenset(
        measure
        for measure in cls.model_fields
        if cls.get_property_info(measure).stereotype == "Measure"
    )

get_name() cached classmethod

Returns the canonical name of the class.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_name(cls) -> str:
    """Returns the canonical name of the class."""
    return cls.__name__

get_polymorphic_fields() cached classmethod

The data type fields that admit more than one class.

A mapping does not say which of them it was built from, and where the classes share their fields the reader cannot tell either, so these are the fields whose values carry their class as datatype when a dump is asked for it.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_polymorphic_fields(cls) -> frozenset[str]:
    """The data type fields that admit more than one class.

    A mapping does not say which of them it was built from, and where the classes
    share their fields the reader cannot tell either, so these are the fields whose
    values carry their class as `datatype` when a dump is asked for it.
    """
    return frozenset(
        name
        for name in cls.model_fields
        if (prop_info := cls.get_property_info(name)).stereotype == "DataType"
        and isinstance(prop_info.typename, list)
    )

get_property_info(name) cached classmethod

Property information.

Parameters:

Name Type Description Default
name str

The property's name.

required

Returns:

Type Description
PropertyInfo

A typed dataclass holding the information.

Raises:

Type Description
AttributeError

The name was not found in the model fields.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_property_info(cls, name: str) -> PropertyInfo:
    """Property information.

    Args:
        name: The property's name.

    Returns:
        A typed dataclass holding the information.

    Raises:
        AttributeError: The name was not found in the model fields.
    """
    try:
        extra_info = _schema_extra(cls.model_fields[name])
    except KeyError:
        raise AttributeError(f"Unknown property: {name}")
    else:
        stereotype: Final = extra_info["stereotype"]
        typename: str | list[str] = extra_info["typename"]
        return PropertyInfo(
            stereotype=stereotype,
            typename=typename,
            list=not extra_info["multiplicity"].endswith("1"),
            nullable=extra_info["multiplicity"].startswith("0"),
            uom=extra_info.get("uom", None),
            enum=_ensure_subclass(
                getattr(cls.appschema().module, str(typename)), BaseEnum
            )
            if stereotype == "Enumeration"
            else None,
            assoc_info=None
            if stereotype != "Association"
            else AssocInfo(
                reverse=extra_info.get("reverseProperty"),
                source_or_target=extra_info.get("sourceOrTarget"),
                dependent_part=extra_info.get("dependent_part"),
            ),
        )

FeatureType

Bases: BaseType

An independently identifiable object: it carries a UUID of its own.

Everything stored as a row in the coretable, held as a member of a FeatureCollection or returned by a repository is one of these. id is declared here so it resolves on a value typed as this class; the generated appschema roots re-declare it with the same annotation and their own json_schema_extra, which is what get_property_info("id") reads.

It also carries everything a data type has no use for: the geometry validator and the hatGenerAttribut machinery - verified over every appschema module, a Geometry field and hatGenerAttribut occur on feature types only.

It is also the only kind that defers its schema build. Pydantic inlines a nested model's schema into its parent, so a class reached only as a field of another one is validated and constructed without a serializer of its own ever being built - it keeps a MockValSer, and anything that serializes such an instance through pydantic_core raises 'MockValSer' object is not an instance of 'SchemaSerializer'. A feature type is never nested: it is always validated directly, through model_factory then model_validate, so deferring it is free. A data type and the value types in appschema.definitions are never anything but nested, so deferring those only cost them their serializer. It is also where the time is - per appschema module, building every feature type costs 513 ms against 19 ms for every data type and 2 ms for all of definitions.

get_geom_field() cached classmethod

Returns the classes geometry field name, if any.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_field(cls) -> str | None:
    """Returns the classes geometry field name, if any."""
    for name in cls.model_fields:
        if cls.get_property_info(name).stereotype == "Geometry":
            return name
    return None

get_geom_srid()

Returns the object's geometry's SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_srid(self) -> int | None:
    """Returns the object's geometry's SRID, if any."""
    return geom.srid if (geom := self._geom()) else None

get_geom_types() cached classmethod

Returns the types of the geometry attribute.

Source code in xmas_core/model/base.py
@classmethod
@functools.cache
def get_geom_types(cls) -> frozenset[type[GeometryType]]:
    """Returns the types of the geometry attribute."""
    if geom_field := cls.get_geom_field():
        geom_annotation = cls.model_fields[geom_field].annotation
        if not (args := get_args(geom_annotation)):
            return frozenset({cast(type[GeometryType], geom_annotation)})
        return frozenset(arg for arg in args if arg is not NoneType)
    return frozenset()

get_geom_wkt()

Returns the object's eWKT geometry's WKT representation withouth SRID, if any.

Source code in xmas_core/model/base.py
def get_geom_wkt(self) -> str | None:
    """Returns the object's eWKT geometry's WKT representation withouth SRID, if any."""
    return geom.wkt if (geom := self._geom()) else None

DataType

Bases: BaseType

A value object nested inside a feature, with no identity of its own.

It reaches a document only as the value of some feature's property, is written and read inline by its owner's encoder and lands in the coretable inside that feature's properties JSON, so it has no id, is never a collection member, and never stands for a GML element, a coretable row or a JSON-FG object of its own. referenzURL / georefURL are declared on data types only, so their validator and serializer live here.

meta

Holds Appschema: what one supported application schema is, and how names resolve in it.

An Appschema wraps one module under xmas_core/model/appschema/ -- one XStandard at one version, produced from the UML model by ShapeChange and datamodel-code-generator -- and carries the metadata that module declares: name, version, namespace and prefix, and model_factory turns a class name into the class.

It lives here rather than in base so that module stays about the BaseType hierarchy alone, and rather than in appschema/__init__.py because that file is generator output.

A schema resolves classes and a class names its schema, so the two depend on each other. meta imports base and never the reverse; the one line back is BaseType.appschema().

Appschema

Bases: BaseModel

Models a supported appschema.

Used to access metadata like name, version and description for an appschema.

During instantion, validation if an appschema is supported occurs, based on the modules in appschema subdirectory.

The instance stores a reference to the appschema's module, which is used in the model_factory method the retrieve appschema classes.

Usage example
# get a featuretype from appschema
appschema = Appschema.from_prefix(prefix="xplan", version="6.0")
plan = appschema.model_factory(name="BP_Plan")
# show list of supported appschemas
Appschema.supported_appschemas()
# get an enum of supported appschemas
SupportedAppschemas = Appschema.enum()
SupportedAppschemas.XPLAN_6_0.value # -> XPlanGML 6_0

code property

The enum code of the appschema.

max_containment_depth cached property

Longest whole -> part chain from a top-level owner, i.e. how deep a plan can nest.

An upper bound on the hop distance between a plan and anything belonging to it, which is what a bounded graph traversal needs in order not to truncate.

It is the longest path, not the shortest: shortest-path distance conflates a type with its instances, and would put XP_TextAbschnitt one hop away because BP_Plan.texte names that type, while an abschnitt referenced only by an object of the plan actually sits three hops out.

The bound is therefore safe but not tight -- most links on the longest chain also hang directly off a Bereich, so no instance need sit that deep -- and it says nothing about features reached through a part shared with another plan, which no depth excludes.

Returns:

Type Description
int

The longest chain length, or 0 for an appschema whose types own nothing.

Raises:

Type Description
RecursionError

A type transitively owns itself, so no finite depth bounds the traversal. Every supported appschema is acyclic.

navigable_roles cached property

Every association role of this appschema that becomes a refs edge.

One row per (source feature type, role, concrete target type). This is the Python side of the navigable_roles_config table the coretable traversal functions read, and scripts/navigable_roles.py renders that table's seed from it -- so the seed is generated rather than a second hand-maintained encoding of the same graph.

Only a FeatureType is stored as its own coretable row and so produces refs edges. Data types (XP_VerbundenerPlan, LP_VorschlagIntegration*, ...) are serialized into the parent's properties JSON, so their roles never become refs rows.

Association typenames are pre-expanded to concrete classes by the generator, so nothing here walks subclasses.

Returns:

Type Description
frozenset[NavigableRole]

The edges as NavigableRole tuples. Unordered: the walk runs over a frozenset of

frozenset[NavigableRole]

role names, so a caller that renders them sorts first.

ownership_edges cached property

The whole -> part containment graph of this appschema, keyed by owner.

One entry per feature type that existentially owns at least one other (an association whose dependent_part is set), mapping it to the types it owns. Association typenames resolve to concrete classes throughout the supported appschemas, so this is the concrete graph, not one that stops at an abstract supertype.

The ownership rows of navigable_roles keyed by owner -- one walk serves both. Ownership is between stored objects, so restricting the walk to FeatureType drops no edge.

Returns:

Type Description
dict[str, frozenset[str]]

Owner feature type name -> the feature type names it existentially owns.

top_level_featuretypes cached property

Names of the top-level owner feature types of this appschema.

A top-level owner existentially owns at least one part (an association whose dependent_part is set) but is itself never the part of another owner: the containment/deletion roots, e.g. BP_Plan. Derived from the appschema's association metadata as the owner-set minus the part-set.

Returns:

Type Description
list[str]

The owner feature type names, sorted.

version property

The version of the appschema in <major>.<minor> format.

enum() cached classmethod

Return an enumeration of supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def enum(cls) -> type[_SupportedAppschema]:
    """Return an enumeration of supported appschemas."""
    members = {
        f"{appschema.prefix.upper()}_{appschema.full_version.major}_{appschema.full_version.minor}": (
            f"{appschema.prefix.upper()}_{appschema.version.replace('.', '_')}",
            appschema,
        )
        for appschema in cls.supported_appschemas()
    }
    # The functional API, called on the metaclass: a checker reads a call on an enum class
    # that defines `__new__` as constructing one member, and types the result `type[Enum]`.
    return cast(
        type[_SupportedAppschema],
        EnumType.__call__(_SupportedAppschema, "SupportedAppschemas", members),
    )

from_enum(code) classmethod

Return an Appschema instance from enum code.

Parameters:

Name Type Description Default
code str

an appschema code, e.g. XPLAN_6_0

required
Source code in xmas_core/model/meta.py
@classmethod
def from_enum(cls, code: str) -> Appschema:
    """Return an Appschema instance from enum code.

    Args:
        code: an appschema code, e.g. `XPLAN_6_0`
    """
    try:
        return cls.enum()[code].appschema
    except KeyError:
        raise NotImplementedError(f"no appschema with code {code!r}") from None

from_module(module_name) cached classmethod

Builds an Appschema instance from module name.

Parameters:

Name Type Description Default
module_name str

the fully qualified module name

required
Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def from_module(cls, module_name: str) -> Appschema:
    """Builds an Appschema instance from module name.

    Args:
        module_name: the fully qualified module name
    """
    try:
        # module_name is APPSCHEMA_MODULE_NAMES-derived or a cls.__module__, never input
        # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
        module = import_module(module_name)
        model_cls: type[RootModel[Any]] = module.Model
    except (ImportError, AttributeError) as exc:
        raise RuntimeError(
            f"Unable to resolve appschema metadata for module {module_name!r}"
        ) from exc

    if not issubclass(model_cls, RootModel):
        raise TypeError(f"expected RootModel, got {model_cls!r}")

    metadata = model_cls.model_fields["root"]

    return cls.model_validate(
        _schema_extra(metadata)
        | {"description": metadata.description, "module": module}
    )

from_namespace(namespace) classmethod

Builds an Appschema instance from the appschema's namespace.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
namespace str

the namespace URI of the appschema

required
Source code in xmas_core/model/meta.py
@classmethod
def from_namespace(cls, namespace: str) -> Appschema:
    """Builds an Appschema instance from the appschema's namespace.

    Might return a compatible appschema version if no exact match is found.

    Args:
        namespace: the namespace URI of the appschema
    """
    uri = AnyUrl(namespace)
    candidates: list[Appschema] = []
    for appschema in cls.supported_appschemas():
        if appschema.namespace_uri == uri:
            return appschema
        elif str(appschema.namespace_uri)[:-1] == namespace[:-1]:
            candidates.append(appschema)
    if candidates:
        return max(candidates)
    raise NotImplementedError(f"no appschema with namespace {namespace!r}")

from_prefix(prefix, version) classmethod

Builds an Appschema instance from the appschema's prefix and version.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
prefix str

the namespace prefix of the appschema, e.g. xplan or xtrasse

required
version str

the version of the appschema; major and minor are required

required
Source code in xmas_core/model/meta.py
@classmethod
def from_prefix(cls, prefix: str, version: str) -> Appschema:
    """Builds an Appschema instance from the appschema's prefix and version.

    Might return a compatible appschema version if no exact match is found.

    Args:
        prefix: the namespace prefix of the appschema, e.g. `xplan` or `xtrasse`
        version: the version of the appschema; major and minor are required
    """
    sem_version = Version.parse(version, optional_minor_and_patch=True)
    versions = []
    candidates = []
    for appschema in filter(
        lambda appschema: appschema.prefix == prefix, cls.supported_appschemas()
    ):
        # if exact match, return
        if (
            sem_version.major == appschema.full_version.major
            and sem_version.minor == appschema.full_version.minor
        ):
            return appschema
        # if versions are compatible (modules minor version higher or equal), add to candidates
        elif sem_version.is_compatible(appschema.full_version):
            candidates.append(appschema)
        else:
            versions.append(
                f"{appschema.full_version.major}.{appschema.full_version.minor}"
            )
    # return appschema with newer minor version, if found
    # TODO: ensure minor version compatibility without version migration, e.g. via model_validator
    # - in GML it's automatic though
    if candidates:
        return max(candidates)
    if versions:
        e = NotImplementedError(
            f"version {version!r} not supported for prefix {prefix!r}"
        )
        e.add_note(f"available versions: {', '.join(sorted(versions))}")
        raise e
    else:
        raise NotImplementedError(f"no appschema with prefix {prefix!r}")

model_factory(name, expect=None)

model_factory(name: str) -> type[FeatureType | DataType]
model_factory(name: str, expect: type[T]) -> type[T]

Factory method for retrieving the corresponding pydantic model representation of a feature class.

Parameters:

Name Type Description Default
name str

name of the feature class or enumeration

required
expect type[BaseType] | None

the base class the result has to derive from. Pass FeatureType where only an identifiable feature will do, so the result is typed as one and a data type raises instead of failing later on a missing id. Defaults to BaseType, which accepts either kind - and since every appschema class is exactly one of the two (pinned by test_every_appschema_class_is_a_featuretype_or_a_datatype), the bare call is typed as type[FeatureType | DataType] rather than the abstract base, so its result can be passed on where either kind is expected.

None

Raises:

Type Description
ValueError

requested class not found, or it is not a subclass of expect.

Returns:

Type Description
type[BaseType]

The concrete class, typed as the requested expect.

Source code in xmas_core/model/meta.py
def model_factory(
    self, name: str, expect: type[BaseType] | None = None
) -> type[BaseType]:
    """Factory method for retrieving the corresponding pydantic model representation of a feature class.

    Args:
        name: name of the feature class or enumeration
        expect: the base class the result has to derive from. Pass `FeatureType` where only
            an identifiable feature will do, so the result is typed as one and a data type
            raises instead of failing later on a missing `id`. Defaults to `BaseType`,
            which accepts either kind - and since every appschema class is exactly one of
            the two (pinned by `test_every_appschema_class_is_a_featuretype_or_a_datatype`),
            the bare call is typed as `type[FeatureType | DataType]` rather than the
            abstract base, so its result can be passed on where either kind is expected.

    Raises:
        ValueError: requested class not found, or it is not a subclass of `expect`.

    Returns:
        The concrete class, typed as the requested `expect`.
    """
    try:
        cls = getattr(self.module, name)
    except AttributeError:
        raise ValueError(f"featuretype {name!r} not found for appschema {self!r}")
    return _ensure_subclass(cls, expect if expect is not None else BaseType)

supported_appschemas() cached classmethod

Return a list of currently supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def supported_appschemas(cls) -> list[Appschema]:
    """Return a list of currently supported appschemas."""
    return [
        cls.from_module(f"{__package__}.appschema.{module_name}")
        for module_name in APPSCHEMA_MODULE_NAMES
    ]

NavigableRole

Bases: NamedTuple

One association edge, in the column order of the navigable_roles_config table.

The table is the SQL side of the same graph: the coretable traversal functions read it to decide which refs edge to follow and which end of it is the existentially dependent part. scripts/navigable_roles.py renders its seed from Appschema.navigable_roles, so the two cannot drift.

Appschema

Bases: BaseModel

Models a supported appschema.

Used to access metadata like name, version and description for an appschema.

During instantion, validation if an appschema is supported occurs, based on the modules in appschema subdirectory.

The instance stores a reference to the appschema's module, which is used in the model_factory method the retrieve appschema classes.

Usage example
# get a featuretype from appschema
appschema = Appschema.from_prefix(prefix="xplan", version="6.0")
plan = appschema.model_factory(name="BP_Plan")
# show list of supported appschemas
Appschema.supported_appschemas()
# get an enum of supported appschemas
SupportedAppschemas = Appschema.enum()
SupportedAppschemas.XPLAN_6_0.value # -> XPlanGML 6_0

code property

The enum code of the appschema.

max_containment_depth cached property

Longest whole -> part chain from a top-level owner, i.e. how deep a plan can nest.

An upper bound on the hop distance between a plan and anything belonging to it, which is what a bounded graph traversal needs in order not to truncate.

It is the longest path, not the shortest: shortest-path distance conflates a type with its instances, and would put XP_TextAbschnitt one hop away because BP_Plan.texte names that type, while an abschnitt referenced only by an object of the plan actually sits three hops out.

The bound is therefore safe but not tight -- most links on the longest chain also hang directly off a Bereich, so no instance need sit that deep -- and it says nothing about features reached through a part shared with another plan, which no depth excludes.

Returns:

Type Description
int

The longest chain length, or 0 for an appschema whose types own nothing.

Raises:

Type Description
RecursionError

A type transitively owns itself, so no finite depth bounds the traversal. Every supported appschema is acyclic.

navigable_roles cached property

Every association role of this appschema that becomes a refs edge.

One row per (source feature type, role, concrete target type). This is the Python side of the navigable_roles_config table the coretable traversal functions read, and scripts/navigable_roles.py renders that table's seed from it -- so the seed is generated rather than a second hand-maintained encoding of the same graph.

Only a FeatureType is stored as its own coretable row and so produces refs edges. Data types (XP_VerbundenerPlan, LP_VorschlagIntegration*, ...) are serialized into the parent's properties JSON, so their roles never become refs rows.

Association typenames are pre-expanded to concrete classes by the generator, so nothing here walks subclasses.

Returns:

Type Description
frozenset[NavigableRole]

The edges as NavigableRole tuples. Unordered: the walk runs over a frozenset of

frozenset[NavigableRole]

role names, so a caller that renders them sorts first.

ownership_edges cached property

The whole -> part containment graph of this appschema, keyed by owner.

One entry per feature type that existentially owns at least one other (an association whose dependent_part is set), mapping it to the types it owns. Association typenames resolve to concrete classes throughout the supported appschemas, so this is the concrete graph, not one that stops at an abstract supertype.

The ownership rows of navigable_roles keyed by owner -- one walk serves both. Ownership is between stored objects, so restricting the walk to FeatureType drops no edge.

Returns:

Type Description
dict[str, frozenset[str]]

Owner feature type name -> the feature type names it existentially owns.

top_level_featuretypes cached property

Names of the top-level owner feature types of this appschema.

A top-level owner existentially owns at least one part (an association whose dependent_part is set) but is itself never the part of another owner: the containment/deletion roots, e.g. BP_Plan. Derived from the appschema's association metadata as the owner-set minus the part-set.

Returns:

Type Description
list[str]

The owner feature type names, sorted.

version property

The version of the appschema in <major>.<minor> format.

enum() cached classmethod

Return an enumeration of supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def enum(cls) -> type[_SupportedAppschema]:
    """Return an enumeration of supported appschemas."""
    members = {
        f"{appschema.prefix.upper()}_{appschema.full_version.major}_{appschema.full_version.minor}": (
            f"{appschema.prefix.upper()}_{appschema.version.replace('.', '_')}",
            appschema,
        )
        for appschema in cls.supported_appschemas()
    }
    # The functional API, called on the metaclass: a checker reads a call on an enum class
    # that defines `__new__` as constructing one member, and types the result `type[Enum]`.
    return cast(
        type[_SupportedAppschema],
        EnumType.__call__(_SupportedAppschema, "SupportedAppschemas", members),
    )

from_enum(code) classmethod

Return an Appschema instance from enum code.

Parameters:

Name Type Description Default
code str

an appschema code, e.g. XPLAN_6_0

required
Source code in xmas_core/model/meta.py
@classmethod
def from_enum(cls, code: str) -> Appschema:
    """Return an Appschema instance from enum code.

    Args:
        code: an appschema code, e.g. `XPLAN_6_0`
    """
    try:
        return cls.enum()[code].appschema
    except KeyError:
        raise NotImplementedError(f"no appschema with code {code!r}") from None

from_module(module_name) cached classmethod

Builds an Appschema instance from module name.

Parameters:

Name Type Description Default
module_name str

the fully qualified module name

required
Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def from_module(cls, module_name: str) -> Appschema:
    """Builds an Appschema instance from module name.

    Args:
        module_name: the fully qualified module name
    """
    try:
        # module_name is APPSCHEMA_MODULE_NAMES-derived or a cls.__module__, never input
        # nosemgrep: python.lang.security.audit.non-literal-import.non-literal-import
        module = import_module(module_name)
        model_cls: type[RootModel[Any]] = module.Model
    except (ImportError, AttributeError) as exc:
        raise RuntimeError(
            f"Unable to resolve appschema metadata for module {module_name!r}"
        ) from exc

    if not issubclass(model_cls, RootModel):
        raise TypeError(f"expected RootModel, got {model_cls!r}")

    metadata = model_cls.model_fields["root"]

    return cls.model_validate(
        _schema_extra(metadata)
        | {"description": metadata.description, "module": module}
    )

from_namespace(namespace) classmethod

Builds an Appschema instance from the appschema's namespace.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
namespace str

the namespace URI of the appschema

required
Source code in xmas_core/model/meta.py
@classmethod
def from_namespace(cls, namespace: str) -> Appschema:
    """Builds an Appschema instance from the appschema's namespace.

    Might return a compatible appschema version if no exact match is found.

    Args:
        namespace: the namespace URI of the appschema
    """
    uri = AnyUrl(namespace)
    candidates: list[Appschema] = []
    for appschema in cls.supported_appschemas():
        if appschema.namespace_uri == uri:
            return appschema
        elif str(appschema.namespace_uri)[:-1] == namespace[:-1]:
            candidates.append(appschema)
    if candidates:
        return max(candidates)
    raise NotImplementedError(f"no appschema with namespace {namespace!r}")

from_prefix(prefix, version) classmethod

Builds an Appschema instance from the appschema's prefix and version.

Might return a compatible appschema version if no exact match is found.

Parameters:

Name Type Description Default
prefix str

the namespace prefix of the appschema, e.g. xplan or xtrasse

required
version str

the version of the appschema; major and minor are required

required
Source code in xmas_core/model/meta.py
@classmethod
def from_prefix(cls, prefix: str, version: str) -> Appschema:
    """Builds an Appschema instance from the appschema's prefix and version.

    Might return a compatible appschema version if no exact match is found.

    Args:
        prefix: the namespace prefix of the appschema, e.g. `xplan` or `xtrasse`
        version: the version of the appschema; major and minor are required
    """
    sem_version = Version.parse(version, optional_minor_and_patch=True)
    versions = []
    candidates = []
    for appschema in filter(
        lambda appschema: appschema.prefix == prefix, cls.supported_appschemas()
    ):
        # if exact match, return
        if (
            sem_version.major == appschema.full_version.major
            and sem_version.minor == appschema.full_version.minor
        ):
            return appschema
        # if versions are compatible (modules minor version higher or equal), add to candidates
        elif sem_version.is_compatible(appschema.full_version):
            candidates.append(appschema)
        else:
            versions.append(
                f"{appschema.full_version.major}.{appschema.full_version.minor}"
            )
    # return appschema with newer minor version, if found
    # TODO: ensure minor version compatibility without version migration, e.g. via model_validator
    # - in GML it's automatic though
    if candidates:
        return max(candidates)
    if versions:
        e = NotImplementedError(
            f"version {version!r} not supported for prefix {prefix!r}"
        )
        e.add_note(f"available versions: {', '.join(sorted(versions))}")
        raise e
    else:
        raise NotImplementedError(f"no appschema with prefix {prefix!r}")

model_factory(name, expect=None)

model_factory(name: str) -> type[FeatureType | DataType]
model_factory(name: str, expect: type[T]) -> type[T]

Factory method for retrieving the corresponding pydantic model representation of a feature class.

Parameters:

Name Type Description Default
name str

name of the feature class or enumeration

required
expect type[BaseType] | None

the base class the result has to derive from. Pass FeatureType where only an identifiable feature will do, so the result is typed as one and a data type raises instead of failing later on a missing id. Defaults to BaseType, which accepts either kind - and since every appschema class is exactly one of the two (pinned by test_every_appschema_class_is_a_featuretype_or_a_datatype), the bare call is typed as type[FeatureType | DataType] rather than the abstract base, so its result can be passed on where either kind is expected.

None

Raises:

Type Description
ValueError

requested class not found, or it is not a subclass of expect.

Returns:

Type Description
type[BaseType]

The concrete class, typed as the requested expect.

Source code in xmas_core/model/meta.py
def model_factory(
    self, name: str, expect: type[BaseType] | None = None
) -> type[BaseType]:
    """Factory method for retrieving the corresponding pydantic model representation of a feature class.

    Args:
        name: name of the feature class or enumeration
        expect: the base class the result has to derive from. Pass `FeatureType` where only
            an identifiable feature will do, so the result is typed as one and a data type
            raises instead of failing later on a missing `id`. Defaults to `BaseType`,
            which accepts either kind - and since every appschema class is exactly one of
            the two (pinned by `test_every_appschema_class_is_a_featuretype_or_a_datatype`),
            the bare call is typed as `type[FeatureType | DataType]` rather than the
            abstract base, so its result can be passed on where either kind is expected.

    Raises:
        ValueError: requested class not found, or it is not a subclass of `expect`.

    Returns:
        The concrete class, typed as the requested `expect`.
    """
    try:
        cls = getattr(self.module, name)
    except AttributeError:
        raise ValueError(f"featuretype {name!r} not found for appschema {self!r}")
    return _ensure_subclass(cls, expect if expect is not None else BaseType)

supported_appschemas() cached classmethod

Return a list of currently supported appschemas.

Source code in xmas_core/model/meta.py
@classmethod
@functools.cache
def supported_appschemas(cls) -> list[Appschema]:
    """Return a list of currently supported appschemas."""
    return [
        cls.from_module(f"{__package__}.appschema.{module_name}")
        for module_name in APPSCHEMA_MODULE_NAMES
    ]

collection

The FeatureCollection, and the reference check every one of them passes.

A collection is what a whole document decodes to and what one encodes from, so it is the type the encodings in xmas_core.codec hand over and the type the processing in xmas_core.processing rewrites. It lives apart from base for that reason: base holds what a single class is -- the appschema registry and the BaseType hierarchy -- and this holds the aggregate over many of them.

Everything a collection promises is enforced by _check_references_and_srs, which runs on every construction: one SRS, one appschema, and every association resolving to a feature of a compatible type inside the same collection. Associations are intra-document by design, so an unresolvable UUID and an external URL are both invalid data; DROP_INVALID_REFS is how a caller reading a document that carries them asks for them to be removed and reported instead of raised.

DROP_INVALID_REFS = 'drop_invalid_refs' module-attribute

Validation context key asking FeatureCollection to drop invalid references.

INVALID_REFS = 'invalid_refs' module-attribute

Validation context key the dropped references are collected under.

FeatureCollection

Bases: BaseModel

Container for features that provides validation of references.

The features are stored in a dictionary with their ID as key and the feature instance as value.

plans property

The collection's plans - its top-level features - by id.

Derived from features on every access rather than stored, so it cannot be passed in and cannot name a plan the collection does not hold, however the features change.

from_features(features, srid, appschema, *, context=None) classmethod

Builds a collection, passing context to the reference check.

The one constructor every repository uses, so the read options a caller can set - DROP_INVALID_REFS above all - reach the validator the same way whichever format was read.

Parameters:

Name Type Description Default
features dict[UUID, FeatureType]

The features, keyed by id.

required
srid int | None

The collection's spatial reference system identifier. None, for a document whose SRS could not be resolved, fails validation.

required
appschema Appschema

The appschema every feature belongs to.

required
context dict[str, Any] | None

Pydantic validation context; see check_references_and_srs.

None

Returns:

Type Description
FeatureCollection

The validated collection.

Source code in xmas_core/model/collection.py
@classmethod
def from_features(
    cls,
    features: dict[UUID, FeatureType],
    srid: int | None,
    appschema: Appschema,
    *,
    context: dict[str, Any] | None = None,
) -> FeatureCollection:
    """Builds a collection, passing `context` to the reference check.

    The one constructor every repository uses, so the read options a caller can set -
    `DROP_INVALID_REFS` above all - reach the validator the same way whichever format
    was read.

    Args:
        features: The features, keyed by id.
        srid: The collection's spatial reference system identifier. `None`, for a
            document whose SRS could not be resolved, fails validation.
        appschema: The appschema every feature belongs to.
        context: Pydantic validation context; see `check_references_and_srs`.

    Returns:
        The validated collection.
    """
    return cls.model_validate(
        {"features": features, "srid": srid, "appschema": appschema},
        context=context,
    )

get_features()

Yields features stored in the collection.

Source code in xmas_core/model/collection.py
def get_features(self) -> Iterator[FeatureType]:
    """Yields features stored in the collection."""
    return (feature for feature in self.features.values())

get_single_plans()

Yields FeatureCollection objects for every plan in the original collection.

Each one holds exactly one plan, which its plans names: a top-level feature type is never the part of another, so no closure reaches a second.

A plan collection is the containment closure of a top-level feature type, derived from the appschema's dependent_part metadata rather than from hardcoded role names: a role marked target points at a part the feature owns, a role marked source points back at its owner. Both directions are followed, since a file may populate only one of them.

A plan is self-contained, so a reference leaving its collection is invalid data and raises when the yielded collection validates its references.

Raises:

Type Description
ValueError

If the appschema declares no dependent_part metadata, so no containment closure can be derived. Every supported appschema declares it.

Source code in xmas_core/model/collection.py
def get_single_plans(self) -> Iterator[FeatureCollection]:
    """Yields FeatureCollection objects for every plan in the original collection.

    Each one holds exactly one plan, which its `plans` names: a top-level feature type is
    never the part of another, so no closure reaches a second.

    A plan collection is the containment closure of a top-level feature type, derived
    from the appschema's `dependent_part` metadata rather than from hardcoded role
    names: a role marked `target` points at a part the feature owns, a role marked
    `source` points back at its owner. Both directions are followed, since a file may
    populate only one of them.

    A plan is self-contained, so a reference leaving its collection is invalid data and
    raises when the yielded collection validates its references.

    Raises:
        ValueError: If the appschema declares no `dependent_part` metadata, so no
            containment closure can be derived. Every supported appschema declares it.
    """
    if not self.appschema.top_level_featuretypes:
        raise ValueError(
            f"appschema {self.appschema.full_name} declares no ownership metadata, "
            "single plans cannot be derived"
        )
    return self._single_plans()

make_copy(with_id_map=False)

Return a copy of the collection with new IDs.

All feature IDs are renewed and respective references are updated.

Parameters:

Name Type Description Default
with_id_map bool

whether to additionally return a map of old IDs to new IDs

False
Source code in xmas_core/model/collection.py
def make_copy(
    self, with_id_map: bool = False
) -> FeatureCollection | tuple[FeatureCollection, dict[UUID, UUID]]:
    """Return a copy of the collection with new IDs.

    All feature IDs are renewed and respective references are updated.

    Args:
        with_id_map: whether to additionally return a map of old IDs to new IDs
    """
    id_map = {key: uuid7() for key in self.features}
    new_features = {}
    for old_feature in self.features.values():
        new_id = id_map[old_feature.id]
        new_feature = old_feature.model_copy(deep=True)
        new_feature.id = new_id
        for assocation in old_feature.get_associations():
            old_value = getattr(old_feature, assocation)
            if isinstance(old_value, UUID):
                new_value = id_map[old_value]
                setattr(new_feature, assocation, new_value)
            elif isinstance(old_value, list):
                new_list = [
                    id_map[item] if isinstance(item, UUID) else item
                    for item in old_value
                ]
                setattr(new_feature, assocation, new_list)
        new_features[new_id] = new_feature

    new_collection = FeatureCollection(
        features=new_features,
        srid=self.srid,
        appschema=self.appschema,
    )
    if with_id_map:
        return new_collection, id_map
    else:
        return new_collection

InvalidReference

Bases: BaseModel

An association reference a collection cannot support.

Either the referenced UUID names no feature in the collection, or the reference is an external URL. Associations are intra-document by design, so neither resolves to a feature and both are unsupported.

FeatureCollection

Bases: BaseModel

Container for features that provides validation of references.

The features are stored in a dictionary with their ID as key and the feature instance as value.

plans property

The collection's plans - its top-level features - by id.

Derived from features on every access rather than stored, so it cannot be passed in and cannot name a plan the collection does not hold, however the features change.

from_features(features, srid, appschema, *, context=None) classmethod

Builds a collection, passing context to the reference check.

The one constructor every repository uses, so the read options a caller can set - DROP_INVALID_REFS above all - reach the validator the same way whichever format was read.

Parameters:

Name Type Description Default
features dict[UUID, FeatureType]

The features, keyed by id.

required
srid int | None

The collection's spatial reference system identifier. None, for a document whose SRS could not be resolved, fails validation.

required
appschema Appschema

The appschema every feature belongs to.

required
context dict[str, Any] | None

Pydantic validation context; see check_references_and_srs.

None

Returns:

Type Description
FeatureCollection

The validated collection.

Source code in xmas_core/model/collection.py
@classmethod
def from_features(
    cls,
    features: dict[UUID, FeatureType],
    srid: int | None,
    appschema: Appschema,
    *,
    context: dict[str, Any] | None = None,
) -> FeatureCollection:
    """Builds a collection, passing `context` to the reference check.

    The one constructor every repository uses, so the read options a caller can set -
    `DROP_INVALID_REFS` above all - reach the validator the same way whichever format
    was read.

    Args:
        features: The features, keyed by id.
        srid: The collection's spatial reference system identifier. `None`, for a
            document whose SRS could not be resolved, fails validation.
        appschema: The appschema every feature belongs to.
        context: Pydantic validation context; see `check_references_and_srs`.

    Returns:
        The validated collection.
    """
    return cls.model_validate(
        {"features": features, "srid": srid, "appschema": appschema},
        context=context,
    )

get_features()

Yields features stored in the collection.

Source code in xmas_core/model/collection.py
def get_features(self) -> Iterator[FeatureType]:
    """Yields features stored in the collection."""
    return (feature for feature in self.features.values())

get_single_plans()

Yields FeatureCollection objects for every plan in the original collection.

Each one holds exactly one plan, which its plans names: a top-level feature type is never the part of another, so no closure reaches a second.

A plan collection is the containment closure of a top-level feature type, derived from the appschema's dependent_part metadata rather than from hardcoded role names: a role marked target points at a part the feature owns, a role marked source points back at its owner. Both directions are followed, since a file may populate only one of them.

A plan is self-contained, so a reference leaving its collection is invalid data and raises when the yielded collection validates its references.

Raises:

Type Description
ValueError

If the appschema declares no dependent_part metadata, so no containment closure can be derived. Every supported appschema declares it.

Source code in xmas_core/model/collection.py
def get_single_plans(self) -> Iterator[FeatureCollection]:
    """Yields FeatureCollection objects for every plan in the original collection.

    Each one holds exactly one plan, which its `plans` names: a top-level feature type is
    never the part of another, so no closure reaches a second.

    A plan collection is the containment closure of a top-level feature type, derived
    from the appschema's `dependent_part` metadata rather than from hardcoded role
    names: a role marked `target` points at a part the feature owns, a role marked
    `source` points back at its owner. Both directions are followed, since a file may
    populate only one of them.

    A plan is self-contained, so a reference leaving its collection is invalid data and
    raises when the yielded collection validates its references.

    Raises:
        ValueError: If the appschema declares no `dependent_part` metadata, so no
            containment closure can be derived. Every supported appschema declares it.
    """
    if not self.appschema.top_level_featuretypes:
        raise ValueError(
            f"appschema {self.appschema.full_name} declares no ownership metadata, "
            "single plans cannot be derived"
        )
    return self._single_plans()

make_copy(with_id_map=False)

Return a copy of the collection with new IDs.

All feature IDs are renewed and respective references are updated.

Parameters:

Name Type Description Default
with_id_map bool

whether to additionally return a map of old IDs to new IDs

False
Source code in xmas_core/model/collection.py
def make_copy(
    self, with_id_map: bool = False
) -> FeatureCollection | tuple[FeatureCollection, dict[UUID, UUID]]:
    """Return a copy of the collection with new IDs.

    All feature IDs are renewed and respective references are updated.

    Args:
        with_id_map: whether to additionally return a map of old IDs to new IDs
    """
    id_map = {key: uuid7() for key in self.features}
    new_features = {}
    for old_feature in self.features.values():
        new_id = id_map[old_feature.id]
        new_feature = old_feature.model_copy(deep=True)
        new_feature.id = new_id
        for assocation in old_feature.get_associations():
            old_value = getattr(old_feature, assocation)
            if isinstance(old_value, UUID):
                new_value = id_map[old_value]
                setattr(new_feature, assocation, new_value)
            elif isinstance(old_value, list):
                new_list = [
                    id_map[item] if isinstance(item, UUID) else item
                    for item in old_value
                ]
                setattr(new_feature, assocation, new_list)
        new_features[new_id] = new_feature

    new_collection = FeatureCollection(
        features=new_features,
        srid=self.srid,
        appschema=self.appschema,
    )
    if with_id_map:
        return new_collection, id_map
    else:
        return new_collection

InvalidReference

Bases: BaseModel

An association reference a collection cannot support.

Either the referenced UUID names no feature in the collection, or the reference is an external URL. Associations are intra-document by design, so neither resolves to a feature and both are unsupported.