datagraphs.Schema

class documentation

class Schema:

Constructor: Schema(name, version)

In-memory representation of a DataGraphs domain model schema.

A Schema instance tracks every change applied to it over its lifetime and exposes change_report to emit a deterministic, net-effect, semantically-annotated changelog relative to the state at construction. Tracking is always on and adds negligible overhead for typical schema sizes.

Every public mutating method is atomic (all-or-nothing): a rollback transaction is opened at the outermost call boundary and replayed on any exception, so a method that raises leaves the schema completely unchanged — never a partial write. The transaction is scoped to the operation's footprint (a shallow class-list snapshot plus a property-granular undo journal), so its cost is proportional to what the operation touches, not to the schema size — building an N-class schema is O(N), not O(N²). Compound mutations (e.g. create_subclass, or any apply_to_subclasses cascade) are covered as a single unit: their inner self-calls share the outer transaction and never open a nested one. Because a rolled-back operation records nothing, change_report never surfaces a change for an operation the caller saw raise.

Static Method	`create_from`	Create a `Schema` from a dictionary.
Method	`__init__`	Create a new empty schema.
Method	`assign_baseclass`	Set or change the parent (base) class for an existing class.
Method	`assign_class_description`	Set or clear the description of a class.
Method	`assign_label_autogen`	Set an auto-generation pattern on the label property of a class.
Method	`assign_label_property`	Designate an existing property as the label property for a class.
Method	`assign_property_orders`	Reorder properties within classes.
Method	`change_report`	Return a net-effect changelog of all changes since construction.
Method	`clone`	Create a deep copy of the schema.
Method	`create_class`	Create a new class in the schema.
Method	`create_property`	Create a new property on a class.
Method	`create_subclass`	Create a subclass that inherits all properties from the parent class.
Method	`delete_class`	Delete a class from the schema.
Method	`delete_property`	Remove a property from a class.
Method	`find_class`	Find a class definition by name.
Method	`find_property`	Find a property by name within a list of property dicts.
Method	`find_subclasses`	Find all direct subclasses of a given class.
Method	`rename_property`	Rename a property.
Method	`to_dict`	Convert the schema to a plain dictionary.
Method	`to_json`	Serialise the schema to a JSON string.
Method	`update_class`	Update a class's name, description, or parent class.
Method	`update_property`	Update an existing property on a class.
Method	`update_schema_metadata`	Update the schema's name, version, and last modified date.
Constant	`ALL_CLASSES`	Undocumented
Property	`classes`	The list of class definitions in the schema.
Property	`version`	The schema version string.
Static Method	`_descendants`	Transitive descendants of baseclass in BFS order, off a prebuilt index.
Static Method	`_is_legacy_format`	Detect whether a schema dict uses the legacy (old) format.
Method	`_apply_metadata`	Mutate the schema's name/version/last-modified — the untracked core.
Method	`_assign_datatype`	Undocumented
Method	`_assign_enum`	Undocumented
Method	`_assign_inverse_of`	Undocumented
Method	`_assign_is_array`	Undocumented
Method	`_assign_is_filterable`	Undocumented
Method	`_assign_is_optional`	Undocumented
Method	`_assign_is_synonym`	Undocumented
Method	`_assign_property_description`	Undocumented
Method	`_children_index`	Build the parent-name -> direct-children-names index in ONE O(C) pass.
Method	`_class_indices`	Build the (name -> class_def) and (parent -> children) indices in ONE pass.
Method	`_create_property_on_class`	Create one property on one already-resolved class dict.
Method	`_delete_linked_properties`	Undocumented
Method	`_get_description_text`	Extract plain text from a description (handles both str and dict).
Method	`_is_valid_inverse_of`	Undocumented
Method	`_make_description`	Create a description dict in the new format.
Method	`_set_internal_schema`	Undocumented
Method	`_transitive_subclass_names`	Names of every transitive subclass of baseclass, in BFS order.
Method	`_update_property_on_class`	Update one already-resolved property on one already-resolved class.
Method	`_validate_schema`	Undocumented
Instance Variable	`_schema`	Undocumented
Instance Variable	`_tracker`	Undocumented
Instance Variable	`_version`	Undocumented

@staticmethod

def create_from(data: dict, version: str = '') -> Self: ¶

Create a Schema from a dictionary.

Automatically detects and converts legacy-format schemas.

Parameters
data:`dict`	Schema dictionary (new or legacy format).
version:`str`	Schema version override.

Returns
`Self`	A new `Schema` instance.

Raises
`SchemaError`	If the dict is missing required keys.

def __init__(self, name: str = '', version: str = ''): ¶

Create a new empty schema.

Parameters
name:`str`	Model name. Defaults to `'Domain Model'` if empty.
version:`str`	Schema version. Defaults to `'1.0'` if empty.

Raises
`TypeError`	If a `dict` is passed instead of keyword arguments.

def assign_baseclass(self, class_name: str, parent_class_name: str): ¶

Set or change the parent (base) class for an existing class.

Parameters
class_name:`str`	The class to modify.
parent_class_name:`str`	The new parent class name.

Raises
`ClassNotFoundError`	If class_name does not exist.

def assign_class_description(self, class_name: str, description: str): ¶

Set or clear the description of a class.

Parameters
class_name:`str`	Class name.
description:`str`	New description. Pass an empty string to remove it.

Raises
`ClassNotFoundError`	If the class does not exist.

def assign_label_autogen(self, class_name: str, pattern: str): ¶

Set an auto-generation pattern on the label property of a class.

Parameters
class_name:`str`	Class name.
pattern:`str`	Auto-generation expression.

Raises
`ClassNotFoundError`	If the class does not exist.
`PropertyNotFoundError`	If the label property does not exist.

def assign_label_property(self, class_name: str, prop_name: str, is_lang_string: bool = True): ¶

Designate an existing property as the label property for a class.

The property is also marked as required (isOptional=False).

Parameters
class_name:`str`	Class name.
prop_name:`str`	Property name to use as the label.
is_lang_string:`bool`	Whether the label supports multiple languages.

Raises
`ClassNotFoundError`	If the class does not exist.
`PropertyNotFoundError`	If the property does not exist on the class.

def assign_property_orders(self, property_orders: dict): ¶

Reorder properties within classes.

Properties not listed in the order are appended at the end.

Parameters
property_orders:`dict`	A dict mapping class names to ordered lists of property names.

def change_report(self, fmt: REPORT_FORMAT = REPORT_FORMAT.TEXT) -> str | list[dict]: ¶

Return a net-effect changelog of all changes since construction.

Computes the structural delta between the baseline (state at construction) and the current schema, then annotates it with semantic intent from the op-log (renames, reorders, compound ops, label-property assignments). The result is deterministic: identical mutation sequences always yield byte-identical text and equal records regardless of dict insertion order.

This method is strictly read-only: it never mutates _schema, _baseline, or _change_log.

Supported surface / guarantees. fmt="records" is the fully-supported, guaranteed output: deterministic and complete — every structural change since the baseline is present, with its full from/to/fields/detail payload, for programmatic consumption. fmt="text" is a best-effort human-readable rendering of the same change set; it is NOT guaranteed to round-trip user-supplied field content (e.g. a description containing newlines may produce additional or ambiguous lines in the text changelog) — a documented known limitation. Cross-subclass annotation of apply_to_subclasses cascade ops in the report is likewise best-effort. Prefer fmt="records" whenever the output is parsed or relied upon.

Cost. For cascade-heavy edit histories the report is approximately O(L*C) (L cascade ops over a parent of C subclasses): a cascade op genuinely fans out to one record per annotated subclass, so the report size — and therefore its cost — is inherent to annotating C subclasses.

Note

Untracked edits via to_dict — graceful degradation. to_dict returns the live internal dict; mutations applied directly to that dict bypass the op-log entirely. Those changes are still captured by the structural diff and appear in change_report output, but without semantic intent labels: a property rename done through the dict appears as a remove + add rather than a single renamed entry, an unlogged reorder does not become a reordered entry, and so on. Use the public mutating methods to preserve full semantic annotation.

Record shape (fmt="records")

Each dict always carries:

"target" (str) — dotted path of the changed entity, using the current name: "ClassName" for class/metadata changes or "ClassName.propName" for property changes.
"kind" (str) — "class", "property", or "metadata".
"op" (str) — one of "added", "removed", "modified", "renamed", "reordered", "subclass_created".

The following keys are omitted (not None) when they do not apply to the entry:

"from" (str) — previous name; present only when op="renamed".
"to" (str) — new name; present only when op="renamed".
"fields" (list[dict]) — field-level before/after list, each entry {"field": str, "before": Any, "after": Any}; present on op="modified" and on op="renamed" when field-level changes accompany the rename.
"detail" (dict) — supplementary annotation dict; present for compound or annotated entries:
- op="subclass_created": {"parent": str, "inherited": int}
- op="reordered": {"order": list[str]}
- op="added" / op="modified" with apply_to_subclasses: {"applied_to_subclasses": list[str]}
- op="modified" (label-property assignment): {"label_property": str}

Parameters

fmt:REPORT_FORMAT

Output format, a ~datagraphs.enums.REPORT_FORMAT. As that enum is a ~enum.StrEnum, the equivalent string ("text" / "records") is also accepted.

~datagraphs.enums.REPORT_FORMAT.TEXT (default): returns a deterministic plain-text changelog str with a header count line and per-class grouping. Best-effort human rendering — see Supported surface above.
~datagraphs.enums.REPORT_FORMAT.RECORDS: returns a list[dict] of structured change records for programmatic consumption — the supported, guaranteed output. See Record shape below.

Returns
`str \| list[dict]`	A `str` for `REPORT_FORMAT.TEXT`; a `list[dict]` for `REPORT_FORMAT.RECORDS`. Returns `""` (text) or `[]` (records) when nothing has changed since construction.

Raises
`ValueError`	If fmt is not a member (or value) of `~datagraphs.enums.REPORT_FORMAT`.

def clone(self) -> Self: ¶

Create a deep copy of the schema.

Returns
`Self`	A new independent `Schema` instance.

def create_class(self, class_name: str, description: str = '', parent_class_name: str = '', label_prop_name: str = 'label', is_label_prop_lang_string: bool = True): ¶

Create a new class in the schema.

Parameters
class_name:`str`	Name of the new class.
description:`str`	Human-readable description.
parent_class_name:`str`	Name of the parent class (for inheritance).
label_prop_name:`str`	Name of the label property created by default.
is_label_prop_lang_string:`bool`	Whether the label property supports multiple languages.

Raises
`SchemaError`	If a class with the same name already exists.

def create_property(self, class_name: str, prop_name: str, datatype: DATATYPE | str, description: str = '', is_optional: bool = True, is_array: bool = False, is_nested: bool = False, is_lang_string: bool = True, inverse_of: str = '', enums: list | None = None, is_synonym: bool = False, is_filterable: bool | None = None, apply_to_subclasses: bool = False): ¶

Create a new property on a class.

Parameters
class_name:`str`	Class to add the property to.
prop_name:`str`	Property name.
datatype:`DATATYPE \| str`	A `DATATYPE` enum value for primitive types, or a class name string for object (relationship) properties.
description:`str`	Human-readable description.
is_optional:`bool`	Whether the property is optional.
is_array:`bool`	Whether the property holds multiple values.
is_nested:`bool`	Whether an object property is nested (embedded).
is_lang_string:`bool`	For text properties, whether to support multiple languages.
inverse_of:`str`	Name of the inverse property on the target class (object properties only).
enums:`list \| None`	Allowed values for `DATATYPE.ENUM` properties.
is_synonym:`bool`	Whether this property is a label synonym.
is_filterable:`bool \| None`	Whether the property is available as a facet/filter.
apply_to_subclasses:`bool`	If `True`, also creates the property on all existing subclasses.

Raises
`ClassNotFoundError`	If the class (or referenced class) does not exist.
`PropertyExistsError`	If a property with the same name already exists.
`InvalidInversePropertyError`	If the inverse property specification is invalid.

def create_subclass(self, class_name: str, description: str, parent_class_name: str): ¶

Create a subclass that inherits all properties from the parent class.

Parameters
class_name:`str`	Name of the new subclass.
description:`str`	Description for the subclass.
parent_class_name:`str`	Name of the parent class to inherit from.

Raises
`ClassNotFoundError`	If the parent class does not exist.

def delete_class(self, class_name: str, include_linked_properties: bool = False, cascade_to_subclasses: bool = True): ¶

Delete a class from the schema.

Parameters
class_name:`str`	Name of the class to delete.
include_linked_properties:`bool`	If `True`, also removes ObjectProperties on other classes that reference this class.
cascade_to_subclasses:`bool`	If `True`, removes `subClassOf` links from any subclasses of the deleted class.

Raises
`ClassNotFoundError`	If the class does not exist.

def delete_property(self, class_name: str, prop_name: str): ¶

Remove a property from a class.

Parameters
class_name:`str`	Class containing the property.
prop_name:`str`	Property name to delete.

Raises
`ClassNotFoundError`	If the class does not exist.
`PropertyNotFoundError`	If the property does not exist.

def find_class(self, name: str) -> dict | None: ¶

Find a class definition by name.

Parameters
name:`str`	The class name to look up.

Returns
`dict \| None`	The class dict, or `None` if not found.

def find_property(self, props: list, name: str) -> dict | None: ¶

Find a property by name within a list of property dicts.

Parameters
props:`list`	List of property dicts to search.
name:`str`	The property name to look up.

Returns
`dict \| None`	The property dict, or `None` if not found.

def find_subclasses(self, baseclass: str) -> list[dict]: ¶

Find all direct subclasses of a given class.

Parameters
baseclass:`str`	The parent class name.

Returns
`list[dict]`	A list of class dicts whose `subClassOf` matches baseclass.

def rename_property(self, class_name: str, old_prop_name: str, new_prop_name: str): ¶

Rename a property.

If the property is the class's label property, the label property reference is updated automatically.

Parameters
class_name:`str`	Class containing the property.
old_prop_name:`str`	Current property name.
new_prop_name:`str`	New property name.

Raises
`ClassNotFoundError`	If the class does not exist.
`PropertyNotFoundError`	If old_prop_name does not exist.
`PropertyExistsError`	If new_prop_name is already in use.

def to_dict(self) -> dict: ¶

Convert the schema to a plain dictionary.

Returns
`dict`	The schema as a dict.

def to_json(self) -> str: ¶

Serialise the schema to a JSON string.

Returns
`str`	A JSON-formatted string.

def update_class(self, class_name: str, new_name: str = '', new_description: str = '', parent_class_name: str = ''): ¶

Update a class's name, description, or parent class.

Parameters
class_name:`str`	Current class name.
new_name:`str`	New class name, or empty to leave unchanged.
new_description:`str`	New description, or empty to leave unchanged.
parent_class_name:`str`	New parent class. Empty string removes the parent.

Raises
`ClassNotFoundError`	If the class does not exist.

Update an existing property on a class.

Only parameters that are explicitly provided (non-None) will be changed.

Parameters
class_name:`str`	Class containing the property.
prop_name:`str`	Property name to update.
datatype:`DATATYPE \| str \| None`	New data type.
description:`str \| None`	New description.
is_optional:`bool \| None`	Whether the property is optional.
is_array:`bool \| None`	Whether the property holds multiple values.
is_nested:`bool \| None`	Whether an object property is nested.
is_lang_string:`bool \| None`	Whether the property supports multiple languages.
inverse_of:`str`	Name of the inverse property on the target class.
enums:`list \| None`	Allowed enumeration values.
is_synonym:`bool`	Whether this property is a label synonym.
is_filterable:`bool \| None`	Whether the property is available as a filter.
apply_to_subclasses:`bool \| None`	If `True`, also updates the property on all existing subclasses.

Raises
`ClassNotFoundError`	If the class does not exist.
`PropertyNotFoundError`	If the property does not exist.

def update_schema_metadata(self, name: str = '', version: str = ''): ¶

Update the schema's name, version, and last modified date.

Parameters
name:`str`	New name for the schema. If empty, the name is unchanged unless it was previously empty, in which case it defaults to 'Domain Model'.
version:`str`	New version string. If empty, the version is unchanged unless it was previously empty, in which case it defaults to '1.0'.

ALL_CLASSES: str = ¶

Undocumented

Value

'__all_classes__'

@property

classes: list[dict] = ¶

The list of class definitions in the schema.

@property

version: str = ¶

The schema version string.

@staticmethod

def _descendants(baseclass: str, children: dict[str, list[str]]) -> list[str]: ¶

Transitive descendants of baseclass in BFS order, off a prebuilt index.

ITERATIVE (explicit queue), so a subClassOf chain thousands of levels deep cannot exceed Python's recursion limit (FIX round-4 B4). Each class is visited at most once (cycle-safe).

@staticmethod

def _is_legacy_format(schema: dict) -> bool: ¶

Detect whether a schema dict uses the legacy (old) format.

def _apply_metadata(self, name: str = '', version: str = ''): ¶

Mutate the schema's name/version/last-modified — the untracked core.

Shared by update_schema_metadata (which wraps this in the tracking + atomic guards and records the op) and by construction paths (__init__ / _set_internal_schema), which apply metadata before a tracker exists and must not record a change.

Parameters
name:`str`	New model name (see `update_schema_metadata`).
version:`str`	New version string (see `update_schema_metadata`).

def _assign_datatype(self, prop_def: dict, datatype: DATATYPE | str, is_nested: bool = False, is_lang_string: bool = True): ¶

Undocumented

def _assign_enum(self, prop_def: dict, datatype: DATATYPE | str, enums: list): ¶

Undocumented

def _assign_inverse_of(self, prop_def: dict, class_name: str, inverse_of: str, datatype: DATATYPE | str): ¶

Undocumented

def _assign_is_array(self, prop_def: dict, is_array: bool = False): ¶

Undocumented

def _assign_is_filterable(self, prop_def: dict, is_filterable: bool | None = None): ¶

Undocumented

def _assign_is_optional(self, prop_def: dict, is_optional: bool = False): ¶

Undocumented

def _assign_is_synonym(self, prop_def: dict, is_synonym: bool): ¶

Undocumented

def _assign_property_description(self, prop_def: dict, description: str): ¶

Undocumented

def _children_index(self) -> dict[str, list[str]]: ¶

Build the parent-name -> direct-children-names index in ONE O(C) pass.

Built once per outermost cascade so the iterative descendant walk is O(descendants) rather than an O(C) find_subclasses scan per level (FIX round-4 B3 — the relocated op-time quadratic).

def _class_indices(self) -> tuple[dict[str, dict], dict[str, list[str]]]: ¶

Build the (name -> class_def) and (parent -> children) indices in ONE pass.

Both indices back the cascade in O(descendants): the name index makes the atomic pre-validation O(targets) (an O(1) lookup per target rather than an O(C) find_class scan), and the children index drives the iterative descendant walk — together eliminating the O(C^2) cascade (FIX round-4 B3).

def _create_property_on_class(self, class_def: dict, owner_class_name: str, prop_name: str, datatype: DATATYPE | str, description: str, is_optional: bool, is_array: bool, is_nested: bool, is_lang_string: bool, inverse_of: str, enums: list, is_synonym: bool, is_filterable: bool | None): ¶

Create one property on one already-resolved class dict.

The single-class core shared by create_property and its cascade. The caller pre-validates existence/duplicate; this core may still raise mid-apply (_assign_datatype on a missing object range, _assign_inverse_of on an invalid inverse) AFTER appending the half-built dict — the caller's outermost _tracker.atomic guard rolls the model back on any such raise, so the overall create is all-or-nothing. inverse_of is resolved against owner_class_name (each target's own class name, matching the prior per-subclass recursion's inverse validation exactly).

def _delete_linked_properties(self, class_name: str): ¶

Undocumented

def _get_description_text(self, desc: str | dict) -> str: ¶

Extract plain text from a description (handles both str and dict).

def _is_valid_inverse_of(self, class_name: str, inverse_of: str, datatype: DATATYPE | str) -> bool: ¶

Undocumented

def _make_description(self, text: str) -> dict: ¶

Create a description dict in the new format.

def _set_internal_schema(self, data: dict, version: str): ¶

Undocumented

def _transitive_subclass_names(self, baseclass: str) -> list[str]: ¶

Names of every transitive subclass of baseclass, in BFS order.

Mirrors the cascade footprint of apply_to_subclasses=True (direct children, their children, and so on). A class is visited at most once (cycle-safe). Builds the children index once and walks it iteratively.

def _update_property_on_class(self, class_def: dict, prop_def: dict, owner_class_name: str, datatype: DATATYPE | str, description, is_optional, is_array, is_nested, is_lang_string, inverse_of, enums, is_synonym, is_filterable): ¶

Update one already-resolved property on one already-resolved class.

The single-class core shared by update_property and its cascade. Only explicitly-provided (non-None) fields are changed, exactly as the public method. inverse_of is resolved against owner_class_name (each target's own class name, matching the prior per-subclass recursion).

def _validate_schema(self, schema: dict): ¶

Undocumented

_schema = ¶

Undocumented

_tracker = ¶

Undocumented

_version = ¶

Undocumented