API Reference¶

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"

class LinkedBaseModel(BaseModel, GenericLinkedBaseModel, metaclass=LinkedBaseModelMetaClass):
    """LinkedBaseModel for pydantic v2"""

    __iris__: dict[str, str | list[str]] | None = {}

    @classmethod
    def get_cls_iri(cls) -> str | list[str] | None:
        """Return the unique IRI of the class.
        Overwrite this method in the subclass."""
        schema = {}
        # pydantic v2
        if hasattr(cls, "model_config") and "json_schema_extra" in cls.model_config:
            schema = cls.model_config["json_schema_extra"]

        cls_iri = []
        # schema annotation - should be expanded IRI
        if "$id" in schema:
            cls_iri.append(schema["$id"])
        elif "iri" in schema:
            cls_iri.append(schema["iri"])
        # default value of type field - may be compacted IRI
        type_field_name = cls.get_type_field()
        # pydantic v2
        type_field = cls.model_fields.get(type_field_name, None)
        if type_field is not None and type_field.default not in cls_iri:
            cls_iri.append(type_field.default)

        if len(cls_iri) == 0:
            return None
        elif len(cls_iri) == 1:
            return cls_iri[0]
        else:
            return cls_iri

    def get_iri(self) -> str:
        """Return the unique IRI of the object.
        Overwrite this method in the subclass."""
        return self.id

    @classmethod
    def model_validate(
        cls,
        obj: Any,
        *,
        strict: bool | None = None,
        from_attributes: bool | None = None,
        context: Any | None = None,
    ) -> Self:
        """Validate a pydantic model instance.

        Args:
            obj: The object to validate.
            strict: Whether to enforce types strictly.
            from_attributes: Whether to extract data from object attributes.
            context: Additional context to pass to the validator.

        Raises:
            ValidationError: If the object could not be validated.

        Returns:
            The validated model instance.
        """
        if isinstance(obj, str):
            return cls._resolve([obj])[obj]
        if isinstance(obj, list):
            node_dict = cls._resolve(obj)
            node_list = []
            for iri in obj:
                node = node_dict[iri]
                if node:
                    node_list.append(node)
            return node_list
        elif isinstance(obj, dict):
            super().model_validate(obj, strict=strict, from_attributes=from_attributes, context=context)

    def __init__(self, *a, **kw):
        # Accept a model instance as first positional arg:
        # TargetModel(source_model, extra_field=value)
        if a and isinstance(a[0], BaseModel):
            source = a[0]
            result = source.cast(type(self), **kw)
            kw = result._raw_dict()
            kw["__iris__"] = getattr(result, "__iris__", {})
            a = ()

        if "__iris__" not in kw:
            kw["__iris__"] = {}

        for name in list(kw):  # force copy of keys for inline-delete
            if name == "__iris__":
                continue
            if name not in self.model_fields:
                continue
            # rewrite <attr> to <attr>_iri
            # pprint(self.__fields__)
            extra = None
            # pydantic v1
            # if name in self.__fields__:
            #     if hasattr(self.__fields__[name].default, "json_schema_extra"):
            #         extra = self.__fields__[name].default.json_schema_extra
            #     elif hasattr(self.__fields__[name].field_info, "extra"):
            #         extra = self.__fields__[name].field_info.extra
            # pydantic v2
            extra = self.model_fields[name].json_schema_extra

            if extra and "range" in extra:
                arg_is_list = isinstance(kw[name], list)

                # annotation_is_list = False
                # args = self.model_fields[name].annotation.__args__
                # if hasattr(args[0], "_name"):
                #    is_list = args[0]._name == "List"
                if arg_is_list:
                    kw["__iris__"][name] = []
                    for e in kw[name][:]:  # interate over copy of list
                        if isinstance(e, BaseModel):  # contructed with object ref
                            kw["__iris__"][name].append(e.get_iri())
                        elif isinstance(e, str):  # constructed from json
                            kw["__iris__"][name].append(e)
                            kw[name].remove(e)  # remove to construct valid instance
                    if len(kw[name]) == 0:
                        # pydantic v1
                        # kw[name] = None # else pydantic v1 will set a FieldInfo object
                        # pydantic v2
                        kw[name] = None  # else default value may be set
                else:
                    if isinstance(kw[name], BaseModel):  # contructed with object ref
                        # print(kw[name].id)
                        kw["__iris__"][name] = kw[name].get_iri()
                    elif isinstance(kw[name], str):  # constructed from json
                        kw["__iris__"][name] = kw[name]
                        # pydantic v1
                        # kw[name] = None # else pydantic v1 will set a FieldInfo object
                        # pydantic v2
                        kw[name] = None  # else default value may be set

        BaseModel.__init__(self, *a, **kw)
        # handle default values
        for name in list(self.__dict__.keys()):
            if self.__dict__[name] is None:
                continue
            extra = None
            # pydantic v1
            # if name in self.__fields__:
            #     if hasattr(self.__fields__[name].default, "json_schema_extra"):
            #         extra = self.__fields__[name].default.json_schema_extra
            #     elif hasattr(self.__fields__[name].field_info, "extra"):
            #         extra = self.__fields__[name].field_info.extra
            # pydantic v2
            extra = self.model_fields[name].json_schema_extra

            if extra and "range" in extra:
                arg_is_list = isinstance(self.__dict__, list)

                if arg_is_list:
                    kw["__iris__"][name] = []
                    for e in self.__dict__[name]:
                        if isinstance(e, BaseModel):  # contructed with object ref
                            kw["__iris__"][name].append(e.get_iri())
                else:
                    if isinstance(self.__dict__[name], BaseModel):  # contructed with object ref
                        kw["__iris__"][name] = self.__dict__[name].get_iri()

        self.__iris__ = kw["__iris__"]

        # iterate over all fields
        # if x-oold-required-iri occurs in extra and the field is not set in __iri__
        # throw an error
        for name in self.model_fields:
            extra = None
            # pydantic v1
            # if name in self.__fields__:
            #     if hasattr(self.__fields__[name].default, "json_schema_extra"):
            #         extra = self.__fields__[name].default.json_schema_extra
            #     elif hasattr(self.__fields__[name].field_info, "extra"):
            #         extra = self.__fields__[name].field_info.extra
            # pydantic v2
            extra = self.model_fields[name].json_schema_extra

            if extra and "x-oold-required-iri" in extra and name not in self.__iris__:
                raise ValueError(f"{name} is required but not set")

    def _handle_value(self, name, value):
        extra = None
        # pydantic v1
        # if name in self.__fields__:
        #     if hasattr(self.__fields__[name].default, "json_schema_extra"):
        #         extra = self.__fields__[name].default.json_schema_extra
        #     elif hasattr(self.__fields__[name].field_info, "extra"):
        #         extra = self.__fields__[name].field_info.extra
        # pydantic v2
        extra = self.model_fields[name].json_schema_extra

        if extra and "range" in extra:
            arg_is_list = isinstance(value, list)

            if arg_is_list:
                self.__iris__[name] = []
                for e in value[:]:  # interate over copy of list
                    if isinstance(e, BaseModel):  # contructed with object ref
                        self.__iris__[name].append(e.get_iri())
                    elif isinstance(e, str):  # constructed from json
                        self.__iris__[name].append(e)
                        value.remove(e)  # remove to construct valid instance
                if len(value) == 0:
                    # pydantic v1
                    value = None  # else pydantic v1 will set a FieldInfo object
                    # pydantic v2
                    # del kw[name]
            else:
                if isinstance(value, BaseModel):  # contructed with object ref
                    # print(value.id)
                    self.__iris__[name] = value.get_iri()
                elif isinstance(value, str):  # constructed from json
                    self.__iris__[name] = value
                    # pydantic v1
                    value = None  # else pydantic v1 will set a FieldInfo object
                    # pydantic v2
                    # del kw[name]
                elif value is None:
                    del self.__iris__[name]
        return value

    def __setattr__(self, name, value, internal=False):
        # print("__setattr__", name, value)
        # Only apply range handling for declared model fields
        if not internal and name in self.model_fields:
            value = self._handle_value(name, value)

        return super().__setattr__(name, value)

    def __getattribute__(self, name):
        # print("__getattribute__ ", name)
        # async? https://stackoverflow.com/questions/33128325/
        # how-to-set-class-attribute-with-await-in-init

        if name in ["__dict__", "__pydantic_private__", "__iris__"]:
            return BaseModel.__getattribute__(self, name)  # prevent loop

        if name == "model_fields":
            return type(self).model_fields

        else:
            if (
                hasattr(self, "__iris__")
                and name in self.__iris__
                and len(self.__iris__[name]) > 0
                and (
                    self.__dict__[name] is None
                    or (isinstance(self.__dict__[name], list) and len(self.__dict__[name]) == 0)
                )
            ):
                iris = self.__iris__[name]
                is_list = isinstance(iris, list)
                if not is_list:
                    iris = [iris]

                node_dict = self._resolve(iris)
                if is_list:
                    node_list = []
                    for iri in iris:
                        node = node_dict[iri]
                        node_list.append(node)
                    self.__setattr__(name, node_list, True)
                else:
                    node = node_dict[iris[0]]
                    if node:
                        self.__setattr__(name, node, True)

        result = BaseModel.__getattribute__(self, name)
        if isinstance(result, list) and name in self.__iris__:
            result = LinkedBaseModelList[type(self)](result, _synced_iri_list=self.__iris__[name])
        return result

    def _raw_dict(self):
        """Serialize to dict without _object_to_iri at any level.

        Used by cast() to preserve inline objects through reconstruction.
        IRI-only fields (value=None, IRI in __iris__) are included as
        IRI strings. Inline objects are recursively serialized as dicts.
        """
        d = {}
        fields = self.model_fields if hasattr(self, "model_fields") else getattr(self, "__fields__", {})
        for name in fields:
            val = self.__dict__.get(name)
            if val is None:
                iri = self.__iris__.get(name) if hasattr(self, "__iris__") else None
                d[name] = iri if iri else None
            elif isinstance(val, list):
                items = []
                for item in val:
                    if hasattr(item, "_raw_dict"):
                        items.append(item._raw_dict())
                    elif hasattr(item, "model_dump"):
                        items.append(item.model_dump())
                    elif hasattr(item, "dict"):
                        items.append(item.dict())
                    else:
                        items.append(item)
                d[name] = items
            elif hasattr(val, "_raw_dict"):
                d[name] = val._raw_dict()
            elif hasattr(val, "model_dump"):
                d[name] = val.model_dump()
            elif hasattr(val, "dict"):
                d[name] = val.dict()
            else:
                d[name] = val
        return d

    def model_dump(self, **kwargs):  # extent BaseClass export function
        # print("dict")
        remove_none = kwargs.get("exclude_none", False)
        kwargs["exclude_none"] = False
        d = super().model_dump(**kwargs)
        # pprint(d)
        self._object_to_iri(d)
        self._recursive_object_to_iri(d, self)
        if remove_none:
            d = self.remove_none(d)
        # pprint(d)
        return d

    @staticmethod
    @staticmethod
    def _recursive_object_to_iri(d: dict, model_obj):
        """Recursively apply __iris__ replacement for nested model objects."""
        fields = model_obj.model_fields if hasattr(model_obj, "model_fields") else getattr(model_obj, "__fields__", {})
        for name, value in list(d.items()):
            if name not in fields:
                continue
            model_value = model_obj.__dict__.get(name)
            if isinstance(value, list) and isinstance(model_value, list):
                for item, model_item in zip(value, model_value, strict=False):
                    if isinstance(item, dict) and hasattr(model_item, "__iris__"):
                        model_item._object_to_iri(item)
                        LinkedBaseModel._recursive_object_to_iri(item, model_item)
            elif isinstance(value, dict) and hasattr(model_value, "__iris__"):
                model_value._object_to_iri(value)
                LinkedBaseModel._recursive_object_to_iri(value, model_value)

    def get_iri_ref(self, field_name: str):
        """Return the stored IRI reference string(s) for a field without
        triggering resolution.

        Parameters
        ----------
        field_name
            The name of the field to retrieve the IRI reference for.

        Returns
        -------
            A string IRI, a list of string IRIs, or ``None`` if no IRI is
            stored for the given field.
        """
        iris = self.__iris__.get(field_name)
        if iris is None:
            return None
        if isinstance(iris, list):
            return iris if iris else None
        return iris

    def get_raw(self, field_name: str):
        """Return the raw value of a field without triggering IRI resolution.

        Unlike normal attribute access which may trigger network calls to
        resolve IRI references, this returns the Python object as stored
        internally (``None`` for unresolved IRIs, the model instance if
        already resolved, or a plain value for non-IRI fields).

        Parameters
        ----------
        field_name
            The name of the field to retrieve.

        Returns
        -------
            The raw field value, or ``None`` if the field is unresolved or
            does not exist.
        """
        return self.__dict__.get(field_name)

    @staticmethod
    def _resolve(iris):
        resolver = get_resolver(GetResolverParam(iri=iris[0])).resolver
        node_dict = resolver.resolve(ResolveParam(iris=iris, model_cls=LinkedBaseModel)).nodes
        return node_dict

    def _store(self):
        backend = get_backend(GetBackendParam(iri=self.get_iri())).backend
        backend.store(StoreParam(nodes={self.get_iri(): self}))

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

    @classmethod
    def _oold_query(cls, query: str | list[str] | Query | Condition) -> "LinkedBaseModelList[Self]":
        # get all resolvers
        # ToDo: filter resolvers that support this class
        resolvers: list[Resolver] = interface._resolvers.values()
        node_list = []
        for r in resolvers:
            try:
                if isinstance(query, (str, list)):
                    _node_list = r.resolve(
                        ResolveParam(
                            iris=[query] if isinstance(query, str) else query,
                            model_cls=cls,
                        )
                    ).nodes.values()
                else:
                    _node_list = r.query(QueryParam(query=query, model_cls=cls)).nodes.values()
                node_list.extend(_node_list)
            except NotImplementedError:
                # resolver does not support query
                continue

        if isinstance(query, str):
            return node_list[0] if len(node_list) > 0 else None
        else:
            return LinkedBaseModelList[Self](node_list, _synced_iri_list=None) if len(node_list) > 0 else None

    @overload
    @classmethod
    def oold_query(cls, item: str) -> Self: ...

    @overload
    @classmethod
    def oold_query(cls, item: list[str]) -> "LinkedBaseModelList[Self]": ...

    # note: (Entity.name == "test") is interpreted as bool
    @overload
    @classmethod
    def oold_query(cls, item: Query | Condition | bool) -> Optional["LinkedBaseModelList[Self]"]: ...

    @classmethod
    def oold_query(
        cls, item: str | list[str] | Query | bool
    ) -> Union[Self, "LinkedBaseModelList[Self]", Optional["LinkedBaseModelList[Self]"]]:
        """Allow access to the class by its IRI."""
        return cls._oold_query(item)
        # if isinstance(item, Query):
        #     # resolve all instances of this class
        #     #print(f"Select all {cls.__name__} that match {index}")
        #     #return cls._oold_query(item)
        #     return cls(id="ex:test", name="test")
        # else:
        #     result = cls._resolve(item if isinstance(item, list) else [item])
        #     return (
        #         result[item] if isinstance(item, str)
        #         else LinkedBaseModelList[Self](
        #             [result[i] for i in item]
        #         )
        #     )

    # pydantic v2
    def model_dump_json(
        self,
        *,
        indent: int | None = None,
        include: pydantic.main.IncEx | None = None,
        exclude: pydantic.main.IncEx | None = None,
        context: Any | None = None,
        by_alias: bool = False,
        exclude_unset: bool = False,
        exclude_defaults: bool = False,
        exclude_none: bool = False,
        round_trip: bool = False,
        warnings: bool | Literal["none", "warn", "error"] = True,
        serialize_as_any: bool = False,
        **dumps_kwargs: Any,
    ) -> str:
        """Usage docs:
        https://docs.pydantic.dev/2.10/concepts/serialization/#modelmodel_dump_json

        Generates a JSON representation of the model using Pydantic's `to_json` method.

        Args:
            indent: Indentation to use in the JSON output.
                If None is passed, the output will be compact.
            include: Field(s) to include in the JSON output.
            exclude: Field(s) to exclude from the JSON output.
            context: Additional context to pass to the serializer.
            by_alias: Whether to serialize using field aliases.
            exclude_unset: Whether to exclude fields that have not been explicitly set.
            exclude_defaults: Whether to exclude fields that are set to
                their default value.
            exclude_none: Whether to exclude fields that have a value of `None`.
            round_trip: If True, dumped values should be valid as input
                for non-idempotent types such as Json[T].
            warnings: How to handle serialization errors. False/"none" ignores them,
                True/"warn" logs errors, "error" raises a
                [`PydanticSerializationError`][pydantic_core.PydanticSerializationError].
            serialize_as_any: Whether to serialize fields with duck-typing serialization
                behavior.

        Returns:
            A JSON string representation of the model.
        """
        d = json.loads(
            BaseModel.model_dump_json(
                self,
                indent=indent,
                include=include,
                exclude=exclude,
                context=context,
                by_alias=by_alias,
                exclude_unset=exclude_unset,
                exclude_defaults=exclude_defaults,
                exclude_none=False,  # handle None values separately
                round_trip=round_trip,
                warnings=warnings,
                serialize_as_any=serialize_as_any,
            )
        )  # ToDo directly use dict?
        # this may replace some None values with IRIs in case they were never resolved
        # thats why we handle exclude_none there
        self._object_to_iri(d)
        self._recursive_object_to_iri(d, self)
        if exclude_none:
            d = self.remove_none(d)
        return json.dumps(d, **dumps_kwargs)

    def cast(
        self,
        cls,
        none_to_default=False,
        remove_extra=False,
        silent=True,
        **kwargs,
    ):
        """Cast this instance to a different model class.

        Parameters
        ----------
        cls
            Target class to cast to.
        none_to_default
            If True, attributes that are None or empty lists are
            removed so the target class uses its defaults.
        remove_extra
            If True, drop fields not defined on the target class.
        silent
            If True, suppress warnings about dropped fields.
        kwargs
            Additional fields to set on the new instance.
        """
        # Use _raw_dict to preserve inline objects at all depths.
        raw = self._raw_dict()
        data = {**raw, **kwargs}
        none_args = []
        if none_to_default:
            reduced = {}
            for k, v in data.items():
                if v is None or (isinstance(v, list) and (len(v) == 0 or all(item is None for item in v))):
                    none_args.append(k)
                else:
                    reduced[k] = v
            data = reduced
        extra_args = []
        if remove_extra:
            target_fields = set()
            if hasattr(cls, "model_fields"):
                target_fields = set(cls.model_fields.keys())
            elif hasattr(cls, "__fields__"):
                target_fields = set(cls.__fields__.keys())
            if target_fields:
                extra_args = [k for k in data if k not in target_fields]
                for k in extra_args:
                    del data[k]
        if not silent:
            if none_to_default and none_args:
                _logger.warning("Removed None/empty attributes: %s", none_args)
            if remove_extra and extra_args:
                _logger.warning("Removed extra attributes: %s", extra_args)
        if "type" in data:
            del data["type"]
        # Carry over __iris__ (range field references)
        if hasattr(self, "__iris__"):
            iris = dict(self.__iris__)
            if "__iris__" in data:
                iris.update(data["__iris__"])
            data["__iris__"] = iris
        return cls(**data)

    def cast_none_to_default(self, cls, **kwargs):
        """Cast to target class, dropping None/empty list attributes."""
        return self.cast(cls, none_to_default=True, **kwargs)

    def to_jsonld(self) -> dict:
        """Return the RDF representation of the object as JSON-LD."""
        return export_jsonld(self, BaseModel)

    @classmethod
    def from_jsonld(cls, jsonld: dict) -> "LinkedBaseModel":
        """Constructs a model instance from a JSON-LD representation."""
        return import_jsonld(BaseModel, LinkedBaseModel, cls, jsonld, _types)

    def to_json(self, exclude_defaults: bool = False) -> dict:
        """Return the JSON representation of the object.

        Parameters
        ----------
        exclude_defaults
            If True, fields with default values are excluded from the
            output. Useful for compact storage where defaults can be
            re-populated on deserialization via from_json().
        """
        result = json.loads(
            self.model_dump_json(
                exclude_none=True,
                exclude_defaults=exclude_defaults,
            )
        )
        # Re-inject IRI-only fields from __iris__ that were excluded
        # because their model value is None (the IRI lives in __iris__)
        if hasattr(self, "__iris__"):
            for field_name, iri in self.__iris__.items():
                if iri is None:
                    continue
                existing = result.get(field_name)
                if existing is None or existing == [] or existing == {}:
                    result[field_name] = iri
        return result

    @classmethod
    def from_json(cls, data: dict) -> "LinkedBaseModel":
        """Constructs a model instance from a JSON representation."""
        return import_json(BaseModel, LinkedBaseModel, cls, data, _types)