Source code for data2rdf.models.base

"""Basic data2rdf models"""

import json
from abc import abstractmethod
from enum import Enum
from typing import Any, Dict, List, Optional, Union

from pydantic import (
    AnyUrl,
    BaseModel,
    ConfigDict,
    Field,
    field_validator,
    model_validator,
)
from rdflib import Graph

from data2rdf.config import Config


[docs]class RelationType(str, Enum): """Relation Type of TBox modellings""" ANNOTATION_PROPERTY = "annotation_property" DATA_PROPERTY = "data_property" OBJECT_PROPERTY = "object_property" PROPERTY = "property"
[docs]class BaseConfigModel(BaseModel): """Basic model for holding the data2rdf config""" config: Config = Field( default_factory=Config, description="Configuration object" ) model_config = ConfigDict(exclude={"config"}) def __str__(self) -> str: """Pretty print the model""" values = ",\n".join( [ f"\t{key}={value}" for key, value in self.__dict__.items() if key not in self.model_config.get("exclude") ] ) return f"{self.__class__.__name__}(\n{values})" def __repr__(self) -> str: """Pretty print the model""" return str(self)
[docs] @field_validator("config") @classmethod def validate_config(cls, value: Union[Dict[str, Any], Config]) -> Config: """Validate configuration""" if isinstance(value, dict): value = Config(**value) return value
[docs]class BasicConceptMapping(BaseConfigModel): """Basic mapping for a concept in a file""" key: Optional[str] = Field( None, description="Key/column of the concept in the file" )
[docs]class BasicGraphModel(BasicConceptMapping): """Basic model for merging data with mappings to become a graph""" @property @abstractmethod def json_ld(self) -> Dict[str, Any]: """Return dict for json-ld of graph""" @property def graph(self) -> Graph: """Return graph object based on json-ld""" graph = Graph(identifier=self.config.graph_identifier) graph.parse(data=json.dumps(self.json_ld), format="json-ld") return graph
[docs]class BasicSuffixModel(BaseConfigModel): """Pydantic BaseModel for suffix and type of a class instance""" iri: Union[str, AnyUrl, List[Union[str, AnyUrl]]] = Field( ..., description="Ontological class related to this concept" ) suffix: Optional[str] = Field( None, description="""Optional suffix of the individual which will be constructed. If not set, the suffix of the iri of the ontological class will be taken""", validate_default=True, )
[docs] @field_validator("iri") @classmethod def validate_iri(cls, value: Union[AnyUrl, List[AnyUrl]]) -> AnyUrl: """Make sure that there are not blank spaces in the IRI""" if not isinstance(value, list): value = AnyUrl(str(value).strip()) else: value = [AnyUrl(str(iterable).strip()) for iterable in value] return value
[docs] @model_validator(mode="after") @classmethod def validate_suffix( cls, self: "BasicSuffixModel", ) -> "BasicSuffixModel": """Return suffix for individal""" if isinstance(self.iri, list) and self.suffix is None: raise TypeError("If the iri is a list, the suffix must be set ") self.suffix = ( self.suffix or str(self.iri).split(self.config.separator)[-1] ) return self