Skip to content

API Reference

Auto-generated from source docstrings via mkdocstrings.


GenericLinkedBaseModel

Base class for all linked data models. Provides JSON-LD serialization, type registry, and the to_jsonld() / to_json() methods.

Source code in src/oold/static.py
class GenericLinkedBaseModel:
    def _object_to_iri(self, d, exclude_none=False):
        for name in list(d.keys()):  # force copy of keys for inline-delete
            if name in self.__iris__:
                d[name] = self.__iris__[name]
            if exclude_none and d[name] is None:
                del d[name]
        return d

    @staticmethod
    def remove_none(d: dict) -> dict:
        """Remove None values from a dictionary recursively."""
        if isinstance(d, dict):
            return {k: GenericLinkedBaseModel.remove_none(v) for k, v in d.items() if v is not None}
        elif isinstance(d, list):
            return [GenericLinkedBaseModel.remove_none(i) for i in d]
        else:
            return d

    @classmethod
    def export_schema(
        cls,
        mode: SchemaExportMode | None = SchemaExportMode.FULL,
        cutoff_base_cls: BaseModel | BaseModel_v1 | tuple[BaseModel | BaseModel_v1] | None = None,
        partial_mode: PartialSchemaExportMode | None = PartialSchemaExportMode.BASE_CLASS_CUTOFF,
        serialize: Literal["json", "yaml"] | None = None,
    ) -> dict:
        """Export the schema of the model as a dictionary."""
        schema = export_schema(cls, mode, cutoff_base_cls, partial_mode)
        if serialize == "json":
            return json.dumps(schema, indent=2)
        elif serialize == "yaml":
            _ignore_aliases = yaml.Dumper.ignore_aliases
            yaml.Dumper.ignore_aliases = lambda *args: True
            yaml_doc = yaml.dump(schema, indent=2)
            yaml.Dumper.ignore_aliases = _ignore_aliases
            return yaml_doc
        return schema

    @classmethod
    @abstractmethod
    def from_jsonld(cls, jsonld: dict) -> "GenericLinkedBaseModel":
        """Constructs a model instance from a JSON-LD representation."""
        pass

    @abstractmethod
    def to_jsonld(self) -> dict:
        """Returns the JSON-LD representation of the model instance as a dictionary."""
        pass

    @classmethod
    @abstractmethod
    def from_json(cls, json_dict: dict) -> "GenericLinkedBaseModel":
        """Constructs a model instance from a JSON representation.
        Note: the given JSON must contain a field to identify the model class,
        default is 'type'."""
        pass

    @abstractmethod
    def to_json(self) -> dict:
        """Return the JSON representation of the object as a dictionary."""
        pass

    @abstractmethod
    def store_jsonld(self):
        """Store the model instance in a backend matching its IRI."""
        pass

    @classmethod
    @abstractmethod
    def get_cls_iri(cls) -> str | list[str] | None:
        """Get the IRI of the model itself.
        It will be used as key for a type registry and should be stored
        in the type field of the JSON(-LD) representation.
        May return both a expanded and a compacted IRI as list of strings."""
        pass

    @classmethod
    def get_type_field(cls) -> str:
        """Get the name of the field that stores the type information.
        It is expected to be aliased or mapped to '@type' in JSON-LD.
        Defaults to 'type'."""
        return "type"

export_schema(mode=SchemaExportMode.FULL, cutoff_base_cls=None, partial_mode=PartialSchemaExportMode.BASE_CLASS_CUTOFF, serialize=None) classmethod

Export the schema of the model as a dictionary.

Source code in src/oold/static.py
@classmethod
def export_schema(
    cls,
    mode: SchemaExportMode | None = SchemaExportMode.FULL,
    cutoff_base_cls: BaseModel | BaseModel_v1 | tuple[BaseModel | BaseModel_v1] | None = None,
    partial_mode: PartialSchemaExportMode | None = PartialSchemaExportMode.BASE_CLASS_CUTOFF,
    serialize: Literal["json", "yaml"] | None = None,
) -> dict:
    """Export the schema of the model as a dictionary."""
    schema = export_schema(cls, mode, cutoff_base_cls, partial_mode)
    if serialize == "json":
        return json.dumps(schema, indent=2)
    elif serialize == "yaml":
        _ignore_aliases = yaml.Dumper.ignore_aliases
        yaml.Dumper.ignore_aliases = lambda *args: True
        yaml_doc = yaml.dump(schema, indent=2)
        yaml.Dumper.ignore_aliases = _ignore_aliases
        return yaml_doc
    return schema

from_json(json_dict) abstractmethod classmethod

Constructs a model instance from a JSON representation. Note: the given JSON must contain a field to identify the model class, default is 'type'.

Source code in src/oold/static.py
@classmethod
@abstractmethod
def from_json(cls, json_dict: dict) -> "GenericLinkedBaseModel":
    """Constructs a model instance from a JSON representation.
    Note: the given JSON must contain a field to identify the model class,
    default is 'type'."""
    pass

from_jsonld(jsonld) abstractmethod classmethod

Constructs a model instance from a JSON-LD representation.

Source code in src/oold/static.py
@classmethod
@abstractmethod
def from_jsonld(cls, jsonld: dict) -> "GenericLinkedBaseModel":
    """Constructs a model instance from a JSON-LD representation."""
    pass

get_cls_iri() abstractmethod classmethod

Get the IRI of the model itself. It will be used as key for a type registry and should be stored in the type field of the JSON(-LD) representation. May return both a expanded and a compacted IRI as list of strings.

Source code in src/oold/static.py
@classmethod
@abstractmethod
def get_cls_iri(cls) -> str | list[str] | None:
    """Get the IRI of the model itself.
    It will be used as key for a type registry and should be stored
    in the type field of the JSON(-LD) representation.
    May return both a expanded and a compacted IRI as list of strings."""
    pass

get_type_field() classmethod

Get the name of the field that stores the type information. It is expected to be aliased or mapped to '@type' in JSON-LD. Defaults to 'type'.

Source code in src/oold/static.py
@classmethod
def get_type_field(cls) -> str:
    """Get the name of the field that stores the type information.
    It is expected to be aliased or mapped to '@type' in JSON-LD.
    Defaults to 'type'."""
    return "type"

remove_none(d) staticmethod

Remove None values from a dictionary recursively.

Source code in src/oold/static.py
@staticmethod
def remove_none(d: dict) -> dict:
    """Remove None values from a dictionary recursively."""
    if isinstance(d, dict):
        return {k: GenericLinkedBaseModel.remove_none(v) for k, v in d.items() if v is not None}
    elif isinstance(d, list):
        return [GenericLinkedBaseModel.remove_none(i) for i in d]
    else:
        return d

store_jsonld() abstractmethod

Store the model instance in a backend matching its IRI.

Source code in src/oold/static.py
@abstractmethod
def store_jsonld(self):
    """Store the model instance in a backend matching its IRI."""
    pass

to_json() abstractmethod

Return the JSON representation of the object as a dictionary.

Source code in src/oold/static.py
@abstractmethod
def to_json(self) -> dict:
    """Return the JSON representation of the object as a dictionary."""
    pass

to_jsonld() abstractmethod

Returns the JSON-LD representation of the model instance as a dictionary.

Source code in src/oold/static.py
@abstractmethod
def to_jsonld(self) -> dict:
    """Returns the JSON-LD representation of the model instance as a dictionary."""
    pass

LinkedBaseModel (v2)

Pydantic v2 implementation. Adds IRI-transparent field resolution, lazy loading, cast(), and the [] subscript operator.

Bases: BaseModel, LinkedApiMixin

Base model supporting both implicit and explicit link declarations.

Source code in src/oold/model/_descriptor.py
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
1350
1351
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
1415
1416
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
1443
1444
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
1477
1478
1479
1480
1481
1482
1483
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
class LinkedBaseModel(BaseModel, LinkedApiMixin, metaclass=LinkedBaseModelMetaClass):
    """Base model supporting both implicit and explicit link declarations."""

    model_config = ConfigDict(ignored_types=(Link, LinkList, _AutoLink))

    _links: dict[str, Any] = PrivateAttr(default_factory=dict)
    # references assigned through __iris__ for names that are not link fields;
    # the shipped side-dict kept them, so reading them back has to work
    _extra_iris: dict[str, Any] = PrivateAttr(default_factory=dict)
    __link_fields__: ClassVar[dict[str, _AutoLink]] = {}
    __link_aliases__: ClassVar[dict[str, str]] = {}
    __required_links__: ClassVar[tuple[str, ...]] = ()
    __link_defaults__: ClassVar[dict[str, Any]] = {}

    @classmethod
    def oold_query(cls, item: Any) -> Any:
        """Resolve ``Model[...]`` against every registered resolver.

        A single IRI yields one instance, a list or a condition yields a list.
        Resolvers that cannot answer a structured query are skipped.
        """
        node_list: list = []
        for resolver in interface._resolvers.values():
            try:
                if isinstance(item, (str, list)):
                    nodes = resolver.resolve(
                        ResolveParam(
                            iris=[item] if isinstance(item, str) else item,
                            model_cls=cls,
                        )
                    ).nodes.values()
                else:
                    nodes = resolver.query(QueryParam(query=item, model_cls=cls)).nodes.values()
                node_list.extend(nodes)
            except NotImplementedError:
                continue
        # A query answers with what it found. An IRI the backend cannot place is
        # not a match, and keeping a None for it would contradict the element
        # type - unlike a to-many link, there is no declaration here promising
        # the result stays aligned with anything.
        node_list = [node for node in node_list if node is not None]
        if isinstance(item, str):
            return node_list[0] if node_list else None
        return LinkResultList(node_list) if node_list else None

    @classmethod
    def __get_pydantic_json_schema__(cls, core_schema_: Any, handler: Any) -> Any:
        """Write ``x-oold-range`` for links that only stated it in the annotation.

        Presence of ``x-oold-range`` is what makes a property a link, so a
        schema carrying only ``x-oold-link`` does not round-trip through code
        generation. ``Link[T]`` names the target already, so the keyword is
        derived from it rather than repeated in ``OoldField(range=...)``.

        Done here, not when the descriptor is installed: a forward reference is
        not resolvable at class-creation time, and a ``Field()`` object shared
        between models must not be mutated in place.
        """
        schema = handler(core_schema_)
        try:
            schema = handler.resolve_ref_schema(schema)
        except Exception:
            return schema
        properties = schema.get("properties") if isinstance(schema, dict) else None
        if not properties:
            return schema
        link_fields = cls.__link_fields__
        aliases = cls.__link_aliases__
        required = schema.get("required")
        for key, prop in properties.items():
            name = key if key in link_fields else aliases.get(key)
            descr = link_fields.get(name) if name else None
            if descr is None or not isinstance(prop, dict):
                continue
            if descr.required_iri:
                # A link is never required at the pydantic level - its value is
                # routed out of the payload before validation - so pydantic
                # leaves it out of `required`. Stating it only in
                # x-oold-required-iri would hide the constraint from every
                # plain JSON Schema validator.
                if required is None:
                    required = schema["required"] = []
                if key not in required:
                    required.append(key)
            if prop.get("x-oold-range") or prop.get("range"):
                continue
            iri = descr.range_iri(cls)
            if iri:
                prop["x-oold-range"] = iri
                # the range says "link" on its own; the marker was a stand-in
                prop.pop("x-oold-link", None)
        return schema

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if not links_enabled():
            # Plain-pydantic mode: install nothing. Link fields keep the
            # semantics their annotation already states - a nested model, not
            # an IRI reference - so the class behaves exactly like a plain
            # BaseModel without touching the declaration.
            cls.__link_fields__ = {}
            return
        links: dict[str, _AutoLink] = dict(getattr(cls, "__link_fields__", {}))
        # Explicit form: descriptors declared directly in the class body.
        for klass in reversed(cls.__mro__):
            for key, value in vars(klass).items():
                if isinstance(value, _AutoLink):
                    links[key] = value
        # Implicit form: annotated fields carrying a range keyword.
        for name, field in cls.model_fields.items():
            extra = field.json_schema_extra
            extra = extra if isinstance(extra, dict) else {}
            rng = extra.get("x-oold-range", extra.get("range"))
            # x-oold-link marks a link whose target comes from the annotation;
            # a Link[...] / LinkList[...] annotation says the same on its own
            if not rng and not extra.get("x-oold-link") and not _is_link_annotation(field.annotation):
                continue
            target, many, optional = _extract_target(field.annotation)
            if isinstance(rng, str) and not isinstance(target, type):
                target = rng
            descr = _AutoLink(name, target, many, optional, bool(extra.get("x-oold-required-iri")))
            setattr(cls, name, descr)
            links[name] = descr
        cls.__link_fields__ = links
        # per-class constants, so construction does not recompute them
        cls.__link_aliases__ = _link_aliases(cls)
        cls.__required_links__ = tuple(n for n, d in links.items() if d.required_iri)
        _register_class(cls)

    def __init__(self, *args: Any, **data: Any) -> None:
        # The shipped model accepts another model as the first positional
        # argument as a cast shorthand: Target(source, extra="value").
        if args and isinstance(args[0], BaseModel):
            source = args[0]
            base = source._raw_dict() if hasattr(source, "_raw_dict") else source.model_dump()
            base.pop("type", None)
            data = {**{k: v for k, v in base.items() if v is not None}, **data}
        elif args:
            raise TypeError(f"{type(self).__name__}() takes no positional arguments other than a source model")
        link_fields = type(self).__link_fields__
        # Route link values out of the payload before pydantic validates - by
        # field name and by alias, since a payload built with by_alias=True uses
        # the alias and would otherwise be validated against the target model.
        aliases = type(self).__link_aliases__
        link_data = {}
        for key in list(data):
            name = key if key in link_fields else aliases.get(key)
            if name is not None:
                link_data[name] = data.pop(key)
        super().__init__(**data)
        # Pydantic writes each field's default into __dict__, and an entry there
        # shadows a non-data descriptor - so an unset link would keep returning
        # that default (None) and never reach __get__. Dropping the entries hands
        # unset links back to the descriptor, which answers [] for to-many and
        # None for to-one. That is what makes a non-Optional list annotation
        # truthful rather than a lie about a value that is really None.
        for _name in link_fields:
            self.__dict__.pop(_name, None)
        # A link field's pydantic default is stripped, so seed the declared IRI
        # here - the link then resolves lazily like any other, instead of the
        # default being lost or fetched on every construction.
        for _name, _iris in type(self).__link_defaults__.items():
            if _name in link_fields and _name not in link_data:
                link_data[_name] = _iris
        for key, value in link_data.items():
            self._set_link(key, value)
        missing = [name for name in type(self).__required_links__ if not self._links.get(name)]
        if missing:
            # x-oold-required-iri, enforced as the legacy binding did. It raised
            # on the mere presence of the keyword; this raises on a true value,
            # so required_iri=False no longer means "required".
            raise ValueError(f"{', '.join(sorted(missing))} is required but not set")

    def _set_link(self, name: str, value: Any) -> None:
        """Store one supplied link value.

        The hook a notation overrides to interpret the value - a union arm has
        to decide literal from reference. Doing it here rather than after
        ``__init__`` returns is what lets the required-link check see the links
        a subclass sets: it ran before them, and reported every required link of
        a notation model as missing.
        """
        type(self).__link_fields__[name].set_value(self, value)

    def __eq__(self, other: Any) -> bool:
        """Compare by data, not by what happens to be cached.

        Resolving a link stores the resolved object in ``__dict__`` (that is
        what makes warm reads native-speed), and pydantic's ``__eq__`` compares
        ``__dict__`` - so reading a link would otherwise change the result of a
        comparison. Links are compared by their stored references instead, and
        the remaining fields the normal way.
        """
        if other.__class__ is not self.__class__:
            return NotImplemented
        # Compare the same state pydantic does - extras and private attributes
        # included. Looking at __dict__ alone made two models with different
        # extra="allow" fields compare equal.
        if self.__pydantic_extra__ != other.__pydantic_extra__:
            return False
        if self.__pydantic_private__ != other.__pydantic_private__:
            return False
        links = type(self).__link_fields__
        if links:
            mine = {k: v for k, v in self.__dict__.items() if k not in links}
            theirs = {k: v for k, v in other.__dict__.items() if k not in links}
            if mine != theirs:
                return False
            return all(links[name].iris(self) == links[name].iris(other) for name in links)
        return self.__dict__ == other.__dict__

    __hash__ = None  # type: ignore[assignment]
    """Unhashable, as pydantic models are.

    An earlier ``__hash__ = id(self)`` made models hashable, so ``set(models)``
    deduplicated by identity instead of raising - silently different from both
    the legacy binding and plain pydantic.
    """

    def __setattr__(self, name: str, value: Any, internal: bool = False) -> None:
        # internal=True means "write the value as given": BaseController passes
        # it through to bypass link handling for controller-only state.
        if name == "__iris__":
            # a property with a setter on the mixin - pydantic would otherwise
            # reject it as "no field __iris__"
            LinkedApiMixin.__iris__.fset(self, value)
            return
        if internal:
            super().__setattr__(name, value)
            return
        # Targeted: only link names are routed to the descriptor. Needed because
        # pydantic's own __setattr__ writes model fields straight into __dict__,
        # bypassing a data descriptor's __set__ (which would leave the link
        # storage and its cache stale). Every other write stays native, and
        # BaseModel already defines __setattr__, so this adds no new slot cost.
        descr = type(self).__link_fields__.get(name)
        if descr is not None:
            descr.set_value(self, value)
        else:
            super().__setattr__(name, value)

    @model_serializer(mode="wrap")
    def _serialize_links(self, handler: Any, info: SerializationInfo) -> dict[str, Any]:
        # Reading a link caches the resolved object in __dict__, where pydantic's
        # own serializer then finds it and serialises it as the declared type.
        # For a to-many link that could not be fully resolved the cache holds a
        # None among the objects, and `list[Bar]` has no way to render it:
        # serialising after such a read died with "type object 'NoneType' has no
        # attribute 'model_fields'". The link keys are replaced below in any
        # case, so the cache is hidden from the handler rather than repaired.
        cached = {}
        for _name in type(self).__link_fields__:
            if _name in self.__dict__:
                cached[_name] = self.__dict__.pop(_name)
        try:
            d = handler(self)
        finally:
            self.__dict__.update(cached)
        fields = type(self).model_fields
        by_alias = bool(getattr(info, "by_alias", False))
        for name, descr in type(self).__link_fields__.items():
            # honour by_alias: every other key does, so writing the link under
            # its field name produced a payload mixing both spellings. The key
            # is never in `d` to compare against - link values are routed out of
            # __dict__ - so the decision comes from the serialisation context.
            name_out = name
            if by_alias:
                field = fields.get(name)
                alias = getattr(field, "serialization_alias", None) or getattr(field, "alias", None)
                if isinstance(alias, str):
                    name_out = alias
            iris = descr.iris(self)
            if iris:
                d.pop(name, None)
                d[name_out] = iris
                continue
            stored = self._links.get(name)
            if stored is None:
                # No value: never set, or explicitly cleared with `= None`.
                # Those are the same statement, and the legacy binding emits
                # neither - so distinguishing them left an explicit null behind
                # after a caller had cleared the link.
                #
                # The key holds None, as the legacy binding does, but only when
                # the caller has not asked for exactly this to be left out.
                # Writing it unconditionally runs *after* handler() has applied
                # the exclusions, which would leak an explicit null past
                # exclude_none, exclude_unset, exclude_defaults and
                # exclude={...} into every stored document.
                d.pop(name, None)
                if not _excluded(info, name):
                    d[name_out] = None
                continue
            # Set, but nothing to reference: either an explicit empty list - a
            # different statement from "unset" and one that must round-trip - or
            # an inline object with no IRI, which has to serialise nested rather
            # than vanish, since cast() is built on this.
            d.pop(name, None)
            d[name_out] = _emit_inline(stored)
        return d

__hash__ = None class-attribute instance-attribute

Unhashable, as pydantic models are.

An earlier __hash__ = id(self) made models hashable, so set(models) deduplicated by identity instead of raising - silently different from both the legacy binding and plain pydantic.

__eq__(other)

Compare by data, not by what happens to be cached.

Resolving a link stores the resolved object in __dict__ (that is what makes warm reads native-speed), and pydantic's __eq__ compares __dict__ - so reading a link would otherwise change the result of a comparison. Links are compared by their stored references instead, and the remaining fields the normal way.

Source code in src/oold/model/_descriptor.py
def __eq__(self, other: Any) -> bool:
    """Compare by data, not by what happens to be cached.

    Resolving a link stores the resolved object in ``__dict__`` (that is
    what makes warm reads native-speed), and pydantic's ``__eq__`` compares
    ``__dict__`` - so reading a link would otherwise change the result of a
    comparison. Links are compared by their stored references instead, and
    the remaining fields the normal way.
    """
    if other.__class__ is not self.__class__:
        return NotImplemented
    # Compare the same state pydantic does - extras and private attributes
    # included. Looking at __dict__ alone made two models with different
    # extra="allow" fields compare equal.
    if self.__pydantic_extra__ != other.__pydantic_extra__:
        return False
    if self.__pydantic_private__ != other.__pydantic_private__:
        return False
    links = type(self).__link_fields__
    if links:
        mine = {k: v for k, v in self.__dict__.items() if k not in links}
        theirs = {k: v for k, v in other.__dict__.items() if k not in links}
        if mine != theirs:
            return False
        return all(links[name].iris(self) == links[name].iris(other) for name in links)
    return self.__dict__ == other.__dict__

__get_pydantic_json_schema__(core_schema_, handler) classmethod

Write x-oold-range for links that only stated it in the annotation.

Presence of x-oold-range is what makes a property a link, so a schema carrying only x-oold-link does not round-trip through code generation. Link[T] names the target already, so the keyword is derived from it rather than repeated in OoldField(range=...).

Done here, not when the descriptor is installed: a forward reference is not resolvable at class-creation time, and a Field() object shared between models must not be mutated in place.

Source code in src/oold/model/_descriptor.py
@classmethod
def __get_pydantic_json_schema__(cls, core_schema_: Any, handler: Any) -> Any:
    """Write ``x-oold-range`` for links that only stated it in the annotation.

    Presence of ``x-oold-range`` is what makes a property a link, so a
    schema carrying only ``x-oold-link`` does not round-trip through code
    generation. ``Link[T]`` names the target already, so the keyword is
    derived from it rather than repeated in ``OoldField(range=...)``.

    Done here, not when the descriptor is installed: a forward reference is
    not resolvable at class-creation time, and a ``Field()`` object shared
    between models must not be mutated in place.
    """
    schema = handler(core_schema_)
    try:
        schema = handler.resolve_ref_schema(schema)
    except Exception:
        return schema
    properties = schema.get("properties") if isinstance(schema, dict) else None
    if not properties:
        return schema
    link_fields = cls.__link_fields__
    aliases = cls.__link_aliases__
    required = schema.get("required")
    for key, prop in properties.items():
        name = key if key in link_fields else aliases.get(key)
        descr = link_fields.get(name) if name else None
        if descr is None or not isinstance(prop, dict):
            continue
        if descr.required_iri:
            # A link is never required at the pydantic level - its value is
            # routed out of the payload before validation - so pydantic
            # leaves it out of `required`. Stating it only in
            # x-oold-required-iri would hide the constraint from every
            # plain JSON Schema validator.
            if required is None:
                required = schema["required"] = []
            if key not in required:
                required.append(key)
        if prop.get("x-oold-range") or prop.get("range"):
            continue
        iri = descr.range_iri(cls)
        if iri:
            prop["x-oold-range"] = iri
            # the range says "link" on its own; the marker was a stand-in
            prop.pop("x-oold-link", None)
    return schema

oold_query(item) classmethod

Resolve Model[...] against every registered resolver.

A single IRI yields one instance, a list or a condition yields a list. Resolvers that cannot answer a structured query are skipped.

Source code in src/oold/model/_descriptor.py
@classmethod
def oold_query(cls, item: Any) -> Any:
    """Resolve ``Model[...]`` against every registered resolver.

    A single IRI yields one instance, a list or a condition yields a list.
    Resolvers that cannot answer a structured query are skipped.
    """
    node_list: list = []
    for resolver in interface._resolvers.values():
        try:
            if isinstance(item, (str, list)):
                nodes = resolver.resolve(
                    ResolveParam(
                        iris=[item] if isinstance(item, str) else item,
                        model_cls=cls,
                    )
                ).nodes.values()
            else:
                nodes = resolver.query(QueryParam(query=item, model_cls=cls)).nodes.values()
            node_list.extend(nodes)
        except NotImplementedError:
            continue
    # A query answers with what it found. An IRI the backend cannot place is
    # not a match, and keeping a None for it would contradict the element
    # type - unlike a to-many link, there is no declaration here promising
    # the result stays aligned with anything.
    node_list = [node for node in node_list if node is not None]
    if isinstance(item, str):
        return node_list[0] if node_list else None
    return LinkResultList(node_list) if node_list else None

BaseController

Mixin for adding runtime behavior to a LinkedBaseModel subclass without polluting the data model or the type registry.

Base mixin for controllers that extend LinkedBaseModel data classes.

Overrides to_json() and to_jsonld() to serialize only the pure data model fields, stripping controller-only fields (e.g. archive_database, auto_archive, connection state).

The data model class is auto-detected from the MRO: the first LinkedBaseModel subclass that is not also a BaseController subclass.

Controllers are excluded from oold's type IRI registry (_types) so they don't replace their pure data model counterparts during backend resolution.

Source code in src/oold/model/__init__.py
class BaseController:
    """Base mixin for controllers that extend LinkedBaseModel data classes.

    Overrides to_json() and to_jsonld() to serialize only the pure data
    model fields, stripping controller-only fields (e.g. archive_database,
    auto_archive, connection state).

    The data model class is auto-detected from the MRO: the first
    LinkedBaseModel subclass that is not also a BaseController subclass.

    Controllers are excluded from oold's type IRI registry (_types) so
    they don't replace their pure data model counterparts during
    backend resolution.
    """

    def __setattr__(self, name, value, internal=False):
        """Route private attrs through object.__setattr__ to bypass
        Pydantic's field validation for controller state fields."""
        if name.startswith("_"):
            object.__setattr__(self, name, value)
        else:
            try:
                super().__setattr__(name, value, internal=internal)
            except (ValueError, AttributeError):
                _logger.warning(
                    "Setting '%s' on %s bypassed Pydantic. Declare it as a Pydantic field or use a _private name.",
                    name,
                    type(self).__name__,
                )
                object.__setattr__(self, name, value)

    def _get_data_model_cls(self):
        """Auto-detect the pure data model class from the MRO.

        Finds all direct LinkedBaseModel bases that are not controllers.
        If there are multiple, creates a dynamic union class combining
        them (e.g. Controller(ModelA, ModelB) -> _ModelA_ModelB).
        """

        def _is_data_model(cls):
            # A data model is recognised by carrying fields, not only by not
            # being on a name list: the descriptor binding mixes in
            # LinkedApiMixin, which answers to_json/from_json but declares no
            # fields, and a name-only test picked it as the data model - so
            # to_json() intersected against an empty field set and returned
            # nothing but the type.
            fields = getattr(cls, "model_fields", None)
            if fields is None:  # pydantic v1 classes
                fields = getattr(cls, "__fields__", None)
            return (
                cls is not type(self)
                and bool(fields)
                and cls.__name__
                not in (
                    "LinkedBaseModel",
                    "_LinkedBaseModel",
                    "_LinkedBaseModelLegacy",
                    "BaseController",
                    "GenericLinkedBaseModel",
                    "BaseModel",
                    "Representation",
                )
                and hasattr(cls, "to_json")
                and hasattr(cls, "from_json")
                and not issubclass(cls, BaseController)
            )

        model_bases = []
        for cls in type(self).__mro__:
            if _is_data_model(cls) and not any(issubclass(m, cls) for m in model_bases):
                model_bases.append(cls)
        if len(model_bases) == 0:
            return None
        if len(model_bases) == 1:
            return model_bases[0]
        name = "_".join(b.__name__ for b in model_bases)
        union_cls = type(name, tuple(model_bases), {})
        union_cls._union_bases = model_bases
        return union_cls

    def _get_model_bases(self):
        """Return the list of pure data model base classes."""
        model_cls = self._get_data_model_cls()
        if model_cls is None:
            return []
        if hasattr(model_cls, "_union_bases"):
            return model_cls._union_bases
        return [model_cls]

    def _collect_type_array(self):
        """Collect merged type array from all pure data model bases."""
        bases = self._get_model_bases()
        if len(bases) <= 1:
            return None
        merged = []
        for base in bases:
            field = None
            if hasattr(base, "model_fields"):
                field = base.model_fields.get("type")
            elif hasattr(base, "__fields__"):
                field = base.__fields__.get("type")
            default = getattr(field, "default", None) if field else None
            if isinstance(default, list):
                for t in default:
                    if t not in merged:
                        merged.append(t)
            elif isinstance(default, str) and default not in merged:
                merged.append(default)
        return merged if merged else None

    def to_json(self, **kwargs):
        # Serialize with LinkedBaseModel.to_json (includes __iris__),
        # then strip controller-only fields
        model_cls = self._get_data_model_cls()
        if model_cls is not None:
            # Use _raw_dict to avoid serialization errors from
            # non-serializable controller fields (e.g. _driver).
            # This bypasses BaseModel.json() which would fail on
            # controller-added fields before we can strip them.
            data = self._raw_dict()
            model_fields = set(
                model_cls.model_fields.keys()
                if hasattr(model_cls, "model_fields")
                else getattr(model_cls, "__fields__", {}).keys()
            )
            for key in list(data.keys()):
                if key not in model_fields and key not in ("type", "@context"):
                    del data[key]
            merged_types = self._collect_type_array()
            if merged_types:
                data["type"] = merged_types
            # Remove None values (match exclude_none behavior)
            data = {k: v for k, v in data.items() if v is not None}
            return data
        return super().to_json(**kwargs)

    def to_jsonld(self):
        data = super().to_jsonld()
        model_cls = self._get_data_model_cls()
        if model_cls is not None:
            merged_types = self._collect_type_array()
            if merged_types:
                data["type"] = merged_types
            model_fields = set(
                model_cls.model_fields.keys()
                if hasattr(model_cls, "model_fields")
                else getattr(model_cls, "__fields__", {}).keys()
            )
            for key in list(data.keys()):
                if key not in model_fields and key not in (
                    "type",
                    "@type",
                    "@context",
                    "@id",
                    "id",
                ):
                    del data[key]
        return data

__setattr__(name, value, internal=False)

Route private attrs through object.setattr to bypass Pydantic's field validation for controller state fields.

Source code in src/oold/model/__init__.py
def __setattr__(self, name, value, internal=False):
    """Route private attrs through object.__setattr__ to bypass
    Pydantic's field validation for controller state fields."""
    if name.startswith("_"):
        object.__setattr__(self, name, value)
    else:
        try:
            super().__setattr__(name, value, internal=internal)
        except (ValueError, AttributeError):
            _logger.warning(
                "Setting '%s' on %s bypassed Pydantic. Declare it as a Pydantic field or use a _private name.",
                name,
                type(self).__name__,
            )
            object.__setattr__(self, name, value)

LinkedBaseModel (v1 - legacy)

Pydantic v1 implementation. Use oold.model.LinkedBaseModel for new projects.

Bases: BaseModel, LinkedApiMixin

pydantic v1 base with the descriptor binding and the downstream API.

Source code in src/oold/model/v1/_descriptor.py
class LinkedBaseModel(BaseModel, LinkedApiMixin, metaclass=LinkedBaseModelMetaClass):
    """pydantic v1 base with the descriptor binding and the downstream API."""

    _links: dict = PrivateAttr(default_factory=dict)
    __link_fields__: dict = {}
    __link_defaults__: dict = {}

    class Config:
        arbitrary_types_allowed = True

    @classmethod
    def oold_query(cls, item: Any) -> Any:
        """Resolve ``Model[...]`` against every registered resolver."""
        from oold.backend import interface
        from oold.backend.interface import QueryParam, ResolveParam

        node_list: list = []
        for resolver in interface._resolvers.values():
            try:
                if isinstance(item, (str, list)):
                    nodes = resolver.resolve(
                        ResolveParam(
                            iris=[item] if isinstance(item, str) else item,
                            model_cls=cls,
                        )
                    ).nodes.values()
                else:
                    nodes = resolver.query(QueryParam(query=item, model_cls=cls)).nodes.values()
                node_list.extend(nodes)
            except NotImplementedError:
                continue
        if isinstance(item, str):
            return node_list[0] if node_list else None
        return LinkResultList(node_list) if node_list else None

    def __init__(self, *args: Any, **data: Any) -> None:
        if args and isinstance(args[0], BaseModel):
            source = args[0]
            base = source._raw_dict() if hasattr(source, "_raw_dict") else source.dict()
            base.pop("type", None)
            data = {**{k: v for k, v in base.items() if v is not None}, **data}
        link_fields = type(self).__link_fields__
        link_data = {k: data.pop(k) for k in list(data) if k in link_fields}
        super().__init__(**data)
        # Pydantic writes each field's default into __dict__, and an entry there
        # shadows a non-data descriptor - so an unset link would keep returning
        # that default (None) and never reach __get__. Dropping the entries hands
        # unset links back to the descriptor, which answers [] for to-many and
        # None for to-one.
        for _name in link_fields:
            self.__dict__.pop(_name, None)
        # seed the declared default IRI, which the neutralisation above took off
        # the field: the link then resolves lazily, like any other
        for _name, _iris in type(self).__link_defaults__.items():
            if _name in link_fields and _name not in link_data:
                link_data[_name] = _iris
        for key, value in link_data.items():
            link_fields[key].set_value(self, value)
        missing = [name for name, d in link_fields.items() if d.required_iri and not self._links.get(name)]
        if missing:
            # see the v2 note: enforced on a true value, not on key presence
            raise ValueError(f"{', '.join(sorted(missing))} is required but not set")

    def __setattr__(self, name: str, value: Any, internal: bool = False) -> None:
        # internal=True means "write the value as given": BaseController passes
        # it through to bypass link handling for controller-only state.
        if name == "__iris__":
            # delegate to the shared property, so a v1 model gets the same
            # replace semantics as a v2 one
            LinkedApiMixin.__iris__.fset(self, value)
            return
        if internal:
            super().__setattr__(name, value)
            return
        descr = type(self).__link_fields__.get(name)
        if descr is not None:
            descr.set_value(self, value)
        else:
            super().__setattr__(name, value)

    # -- downstream API -----------------------------------------------------

    @property
    def __iris__(self) -> dict[str, Any]:
        out: dict[str, Any] = {}
        for name, descr in type(self).__link_fields__.items():
            iris = descr.iris(self)
            if iris:
                out[name] = iris
        return out

    @classmethod
    def _fields(cls) -> dict:
        return cls.__fields__

    def _dump(self, **kwargs: Any) -> dict:
        return self.dict(**kwargs)

    @classmethod
    def get_type_field(cls) -> str:
        return "type"

    @classmethod
    def get_cls_iri(cls) -> Any:
        """The class IRI(s), from ``Config.schema_extra`` and the type default."""
        schema = getattr(getattr(cls, "__config__", None), "schema_extra", None) or {}
        if callable(schema):
            schema = {}
        out: list[str] = []
        for key in ("$id", "x-oold-iri", "iri"):
            if key in schema:
                out.append(schema[key])
                break
        type_field = cls.__fields__.get(cls.get_type_field())
        if type_field is not None:
            default = type_field.default
            for value in default if isinstance(default, list) else [default]:
                if isinstance(value, str) and value not in out:
                    out.append(value)
        if not out:
            return None
        return out[0] if len(out) == 1 else out

    def dict(self, **kwargs: Any) -> dict[str, Any]:
        """v1 serialisation; link fields collapse to their IRIs."""
        exclude_none = kwargs.pop("exclude_none", False)
        links = type(self).__link_fields__
        # Reading a link caches the resolved value in __dict__, which pydantic v1
        # serialises - so whether a link had been read changed the output. Drop
        # the cache entries for the duration, then restore them.
        cached = {name: self.__dict__.pop(name) for name in links if name in self.__dict__}
        try:
            d = super().dict(**kwargs)
        finally:
            self.__dict__.update(cached)
        for name, descr in links.items():
            iris = descr.iris(self)
            if iris:
                d[name] = iris
            else:
                d[name] = None
        if exclude_none:
            d = {k: v for k, v in d.items() if v is not None}
        return d

    def json(self, **kwargs: Any) -> str:
        # dict() leaves UUIDs, datetimes and enums as Python objects, so the
        # model's own encoder has to do the conversion - plain json.dumps
        # rejects them.
        encoder = kwargs.pop("encoder", None) or self.__json_encoder__
        kwargs.pop("models_as_dict", None)
        return json.dumps(self.dict(**kwargs), default=encoder)

    def to_json(self, exclude_defaults: bool = False) -> dict[str, Any]:
        return json.loads(self.json(exclude_none=True, exclude_defaults=exclude_defaults))

    @classmethod
    def from_json(cls, data: dict[str, Any]) -> Any:
        from oold.static import import_json

        return import_json(BaseModel, LinkedBaseModel, cls, data, _TYPE_REGISTRY)

    def to_jsonld(self) -> dict[str, Any]:
        from oold.static import export_jsonld

        return export_jsonld(self, BaseModel)

    @classmethod
    def from_jsonld(cls, jsonld: dict[str, Any]) -> Any:
        from oold.static import import_jsonld

        return import_jsonld(BaseModel, LinkedBaseModel, cls, jsonld, _TYPE_REGISTRY)

dict(**kwargs)

v1 serialisation; link fields collapse to their IRIs.

Source code in src/oold/model/v1/_descriptor.py
def dict(self, **kwargs: Any) -> dict[str, Any]:
    """v1 serialisation; link fields collapse to their IRIs."""
    exclude_none = kwargs.pop("exclude_none", False)
    links = type(self).__link_fields__
    # Reading a link caches the resolved value in __dict__, which pydantic v1
    # serialises - so whether a link had been read changed the output. Drop
    # the cache entries for the duration, then restore them.
    cached = {name: self.__dict__.pop(name) for name in links if name in self.__dict__}
    try:
        d = super().dict(**kwargs)
    finally:
        self.__dict__.update(cached)
    for name, descr in links.items():
        iris = descr.iris(self)
        if iris:
            d[name] = iris
        else:
            d[name] = None
    if exclude_none:
        d = {k: v for k, v in d.items() if v is not None}
    return d

get_cls_iri() classmethod

The class IRI(s), from Config.schema_extra and the type default.

Source code in src/oold/model/v1/_descriptor.py
@classmethod
def get_cls_iri(cls) -> Any:
    """The class IRI(s), from ``Config.schema_extra`` and the type default."""
    schema = getattr(getattr(cls, "__config__", None), "schema_extra", None) or {}
    if callable(schema):
        schema = {}
    out: list[str] = []
    for key in ("$id", "x-oold-iri", "iri"):
        if key in schema:
            out.append(schema[key])
            break
    type_field = cls.__fields__.get(cls.get_type_field())
    if type_field is not None:
        default = type_field.default
        for value in default if isinstance(default, list) else [default]:
            if isinstance(value, str) and value not in out:
                out.append(value)
    if not out:
        return None
    return out[0] if len(out) == 1 else out

oold_query(item) classmethod

Resolve Model[...] against every registered resolver.

Source code in src/oold/model/v1/_descriptor.py
@classmethod
def oold_query(cls, item: Any) -> Any:
    """Resolve ``Model[...]`` against every registered resolver."""
    from oold.backend import interface
    from oold.backend.interface import QueryParam, ResolveParam

    node_list: list = []
    for resolver in interface._resolvers.values():
        try:
            if isinstance(item, (str, list)):
                nodes = resolver.resolve(
                    ResolveParam(
                        iris=[item] if isinstance(item, str) else item,
                        model_cls=cls,
                    )
                ).nodes.values()
            else:
                nodes = resolver.query(QueryParam(query=item, model_cls=cls)).nodes.values()
            node_list.extend(nodes)
        except NotImplementedError:
            continue
    if isinstance(item, str):
        return node_list[0] if node_list else None
    return LinkResultList(node_list) if node_list else None

Backend interface

Abstract interface implemented by all backends.

Bases: Resolver

Source code in src/oold/backend/interface.py
class Backend(Resolver):
    def store(self, param: StoreParam) -> StoreResult:
        jsonld_dicts = {}
        for iri, node in param.nodes.items():
            if node is None:
                jsonld_dicts[iri] = None
            else:
                if self.format == LinkedDataFormat.JSON_LD:
                    jsonld_dicts[iri] = node.to_jsonld()
                elif self.format == LinkedDataFormat.JSON:
                    jsonld_dicts[iri] = node.to_json()
                else:
                    raise ValueError(f"Unsupported format {self.format}")
        if self.format == LinkedDataFormat.JSON:
            return self.store_json_dicts(jsonld_dicts)
        else:
            return self.store_jsonld_dicts(jsonld_dicts)

    def store_jsonld_dicts(self, jsonld_dicts: dict[str, dict]) -> StoreResult:
        raise NotImplementedError("store_jsonld_dicts method not implemented in Backend subclass")

    def store_json_dicts(self, json_dicts: dict[str, dict]) -> StoreResult:
        raise NotImplementedError("store_json_dicts method not implemented in Backend subclass")

Generator

Code generation from OO-LD / JSON Schema definitions.

Source code in src/oold/generator.py
class Generator:
    class GenerateParams(BaseModel):
        json_schemas: list[dict]
        """JSON SCHEMA source(s)"""
        preprocess: bool = True
        """Preprocess the JSON schemas before generating the models"""
        main_schema: str | None = None
        """File name of the main schema"""
        output_model_type: DataModelType | None = (DataModelType.PydanticV2BaseModel,)
        """Output model type, e.g. PydanticV2BaseModel or PydanticBaseModel"""
        output_model_path: Path | None = Path(__file__).parent / "model" / "example.py"
        """Output model path, if not set the model will be generated
        in the current directory"""
        working_dir_path: Path | None = None
        """Working directory to store intermedia files
        and the generated partial models"""
        generate_init_py_files: bool = True
        """Generate __init__.py files along the output_model_path"""

    def generate(
        self,
        params: GenerateParams,
    ):
        if params.preprocess:
            self.preprocess(Generator.PreprocessParams(json_schemas=params.json_schemas))

        # monkey patch class
        datamodel_code_generator.parser.jsonschema.JsonSchemaParser = OOLDJsonSchemaParser

        with TemporaryDirectory() as temporary_directory_name:
            temporary_directory = Path(temporary_directory_name)
            if params.working_dir_path is not None:
                temporary_directory = params.working_dir_path

            input = Path(temporary_directory)
            if params.main_schema is not None:
                input = Path(temporary_directory / Path(params.main_schema))

            output = params.output_model_path
            if params.generate_init_py_files:
                # generate __init__.py files in every subdirectory
                # of the output model path that does not exist yet
                # output may be a file or a directory

                # check if output is a file or a directory
                target_dir = output
                if params.main_schema is not None:
                    target_dir = output.parent

                # interate over the target_dir path, starting at the top level dir
                # e.g. 'C:' or '/var'
                for segment in target_dir.parts:
                    # create the segment path
                    segment_path = Path(*target_dir.parts[: target_dir.parts.index(segment) + 1])
                    # check if the segment path exists
                    if not segment_path.exists():
                        # create the __init__.py file
                        os.makedirs(segment_path, exist_ok=False)
                        init_file = segment_path / "__init__.py"
                        with open(init_file, "w", encoding="utf-8") as f:
                            f.write("# Generated by oold.generator\n")

            for schema in params.json_schemas:
                name = schema["id"]
                os.makedirs(
                    os.path.dirname(Path(temporary_directory / (name + ".json"))),
                    exist_ok=True,
                )
                with open(Path(temporary_directory / (name + ".json")), "w", encoding="utf-8") as f:
                    schema_str = json.dumps(schema, ensure_ascii=False, indent=2).replace("dollarref", "$ref")
                    # print(schema_str)
                    f.write(schema_str + "\n")

            if params.output_model_type == DataModelType.PydanticV2BaseModel:
                base_class = "oold.model.LinkedBaseModel"
            else:
                base_class = "oold.model.v1.LinkedBaseModel"
            generate(
                input_=input,
                # json_schema,
                input_file_type=InputFileType.JsonSchema,
                # input_filename="Foo.json",
                output=output,
                # set up the output model types
                output_model_type=params.output_model_type,
                # custom_template_dir=Path(model_dir_path),
                field_include_all_keys=True,
                base_class=base_class,
                # use_default = True,
                allof_class_hierarchy=(datamodel_code_generator.AllOfClassHierarchy.Always),
                enum_field_as_literal=datamodel_code_generator.LiteralType.Off,
                use_title_as_name=True,
                use_schema_description=True,
                use_field_description=True,
                encoding="utf-8",
                use_double_quotes=True,
                disable_timestamp=True,
                collapse_root_models=True,
                reuse_model=True,
                # create MyEnum(str, Enum) instead of MyEnum(Enum)
                use_subclass_enum=True,
                additional_imports=["pydantic.ConfigDict"]
                if params.output_model_type == DataModelType.PydanticV2BaseModel
                else [],
                apply_default_values_for_required_fields=True,
            )

            if params.main_schema is not None:
                content = ""
                with open(output, encoding="utf-8") as f:
                    content = f.read()
                os.remove(output)

                content = re.sub(
                    r"(UUID = Field\(...)",
                    r"UUID = Field(default_factory=uuid4",
                    content,
                )  # enable default value for uuid

                if params.output_model_type == DataModelType.PydanticBaseModel:
                    # we are now using pydantic.v1
                    # pydantic imports lead to uninitialized fields
                    # (FieldInfo still present)
                    content = re.sub(r"(from pydantic import)", "from pydantic.v1 import", content)

                # fix unserializable defaults from datamodel-code-generator
                # when allOf merges a property override (e.g. hidden:true) with
                # a parent field typed as a complex model, the default becomes
                # an unserializable sentinel: lambda :Foo.parse_obj(<object ...>)
                content = re.sub(
                    r"default_factory=lambda\s*:.*<object object at 0x[0-9a-fA-F]+>\)",
                    "default=None)",
                    content,
                )
                # fix lambda formatting (space before colon breaks black)
                content = content.replace("lambda :", "lambda:")

                # write the content to the file
                with open(output, "w", encoding="utf-8") as f:
                    f.write(content)

    class PreprocessParams(BaseModel):
        json_schemas: list[dict]
        """JSON SCHEMA source(s)"""

    def preprocess(self, params: PreprocessParams):
        for schema in params.json_schemas:
            # schema = self.merge_property_schemas(schema)
            for property_key in schema.get("properties", {}):
                property = schema["properties"][property_key]
                if "range" in property:
                    if "type" in property:
                        del property["type"]
                    # if range is a string we create a allOf with a ref to the range
                    if isinstance(property["range"], str):
                        property["allOf"] = [{"$ref": property["range"]}]
                    else:
                        property["$ref"] = property["range"]
                    if "required" in schema and property_key in schema["required"]:
                        # if no default value is set, remove the property from required
                        if "default" not in property:
                            schema["required"].remove(property_key)
                        if "x-oold-required-iri" not in property:
                            property["x-oold-required-iri"] = True
                if "items" in property:
                    if "range" in property["items"]:
                        if "type" in property["items"]:
                            del property["items"]["type"]
                        if isinstance(property["items"]["range"], str):
                            property["items"]["allOf"] = [{"$ref": property["items"]["range"]}]
                        else:
                            property["items"]["$ref"] = property["items"]["range"]
                        property["range"] = property["items"]["range"]
                        if "required" in schema and property_key in schema["required"]:
                            # if no default value is set,
                            # remove the property from required
                            if "default" not in property["items"]:
                                schema["required"].remove(property_key)
                            if "x-oold-required-iri" not in property:
                                property["x-oold-required-iri"] = True

                    if "properties" in property["items"]:
                        self.preprocess(Generator.PreprocessParams(json_schemas=[property["items"]]))

                if "properties" in property:
                    self.preprocess(Generator.PreprocessParams(json_schemas=[property]))

GenerateParams

Bases: BaseModel

Source code in src/oold/generator.py
class GenerateParams(BaseModel):
    json_schemas: list[dict]
    """JSON SCHEMA source(s)"""
    preprocess: bool = True
    """Preprocess the JSON schemas before generating the models"""
    main_schema: str | None = None
    """File name of the main schema"""
    output_model_type: DataModelType | None = (DataModelType.PydanticV2BaseModel,)
    """Output model type, e.g. PydanticV2BaseModel or PydanticBaseModel"""
    output_model_path: Path | None = Path(__file__).parent / "model" / "example.py"
    """Output model path, if not set the model will be generated
    in the current directory"""
    working_dir_path: Path | None = None
    """Working directory to store intermedia files
    and the generated partial models"""
    generate_init_py_files: bool = True
    """Generate __init__.py files along the output_model_path"""

generate_init_py_files = True class-attribute instance-attribute

Generate init.py files along the output_model_path

json_schemas instance-attribute

JSON SCHEMA source(s)

main_schema = None class-attribute instance-attribute

File name of the main schema

output_model_path = Path(__file__).parent / 'model' / 'example.py' class-attribute instance-attribute

Output model path, if not set the model will be generated in the current directory

output_model_type = (DataModelType.PydanticV2BaseModel,) class-attribute instance-attribute

Output model type, e.g. PydanticV2BaseModel or PydanticBaseModel

preprocess = True class-attribute instance-attribute

Preprocess the JSON schemas before generating the models

working_dir_path = None class-attribute instance-attribute

Working directory to store intermedia files and the generated partial models

PreprocessParams

Bases: BaseModel

Source code in src/oold/generator.py
class PreprocessParams(BaseModel):
    json_schemas: list[dict]
    """JSON SCHEMA source(s)"""

json_schemas instance-attribute

JSON SCHEMA source(s)