Skip to content

OTE-API Configuration Models

This page provides documentation for the oteapi.models submodule, where all the OTE-API strategy configuration data models are located.

When creating instances of these models, the data types are automatically validated.

oteapi.models module.

This module contains all the pydantic configuration models.

HostlessAnyUrl = Annotated[Url, UrlConstraints(host_required=False)] module-attribute

RDFTriple = Tuple[str, str, str] module-attribute

StrategyConfig = Union[FilterConfig, FunctionConfig, MappingConfig, ParserConfig, ResourceConfig, TransformationConfig] module-attribute

AttrDict

Bases: BaseModel, MutableMapping

An object whose attributes can also be accessed through subscription, like with a dictionary.

Special pydantic configuration settings:

  • extra Allow any attributes/fields to be defined - this is what makes this pydantic model an attribute dictionary.
  • validate_assignment Validate and cast set values. This is mainly relevant for sub-classes of AttrDict, where specific attributes have been defined.
  • arbitrary_types_allowed If a custom type is used for an attribute that doesn't have a validate() method, don't fail setting the attribute.
Source code in oteapi/models/genericconfig.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class AttrDict(BaseModel, MutableMapping):
    """An object whose attributes can also be accessed through
    subscription, like with a dictionary.

    Special pydantic configuration settings:

    - **`extra`**
      Allow any attributes/fields to be defined - this is what makes this pydantic
      model an attribute dictionary.
    - **`validate_assignment`**
      Validate and cast set values.
      This is mainly relevant for sub-classes of `AttrDict`, where specific
      attributes have been defined.
    - **`arbitrary_types_allowed`**
      If a custom type is used for an attribute that doesn't have a `validate()`
      method, don't fail setting the attribute.

    """

    model_config = ConfigDict(
        extra="allow", validate_assignment=True, arbitrary_types_allowed=True
    )

    # Collection methods
    def __contains__(self, key: object) -> bool:
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        return hasattr(self, key)

    def __len__(self) -> int:
        return len(self.model_dump())

    # Mapping methods
    def __getitem__(self, key: "Any") -> "Any":
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            return getattr(self, key)
        except AttributeError as exc:
            raise KeyError(key) from exc

    def items(self):
        return self.model_dump().items()

    def keys(self):
        return self.model_dump().keys()

    def values(self):
        return self.model_dump().values()

    def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
        return getattr(self, key, default)

    def __eq__(self, value: object) -> bool:
        if isinstance(value, Mapping):
            return self.model_dump() == value
        if isinstance(value, BaseModel):
            return BaseModel.__eq__(self, value)
        return False

    # MutableMapping methods
    def __setitem__(self, key: "Any", value: "Any") -> None:
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            return self.__setattr__(key, value)
        except AttributeError as exc:
            raise KeyError(key) from exc

    def __delitem__(self, key: "Any") -> None:
        warnings.warn(
            "Item deletion used to reset fields to their default values. To keep using"
            " this functionality, use the `reset_field()` method.",
            DeprecationWarning,
        )

        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            self.__delattr__(key)
        except AttributeError as exc:
            raise KeyError(key) from exc
        else:
            if key in self.model_fields_set:
                self.model_fields_set.remove(key)

    def clear(self) -> None:
        # Ignore the deprecation warning from `__delitem__`
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)

            for field in self.model_dump():
                del self[field]

    def update(  # type: ignore[override]
        self,
        other: "Optional[Union[Mapping[str, Any], Iterable[tuple[str, Any]]]]" = None,
        **kwargs,
    ) -> None:
        if other and isinstance(other, Mapping):
            for key, value in other.items():
                setattr(self, key, value)
        elif other and isinstance(other, BaseModel):
            for key, value in other:
                setattr(self, key, value)
        elif other and isinstance(other, Iterable):
            for entry in other:
                if not isinstance(entry, tuple):
                    raise TypeError(
                        "`other` must be an iterable of tuples of length two."
                    )
                if not len(entry) == 2:
                    raise ValueError(
                        "`other` must be an iterable of objects of length two."
                    )
            for key, value in other:  # type: ignore[misc]
                setattr(self, key, value)
        elif other:
            raise TypeError(
                "`other` must be of type `dict`, `Mapping`, `BaseModel` or "
                "`Iterable`, not `{type(other).__name__}`."
            )
        if kwargs:
            for key, value in kwargs.items():
                setattr(self, key, value)

    def pop(self, key: str, default: "Optional[Any]" = PydanticUndefined) -> "Any":
        value = self.get(key, default)
        if value == PydanticUndefined:
            raise KeyError(key)
        if key in self:
            # Ignore the deprecation warning from `__delitem__`
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", DeprecationWarning)

                del self[key]
        return value

    def popitem(self) -> "Tuple[str, Any]":
        """MutableMapping `popitem`-method.

        Important:
            Unlike the regular `dict.popitem()` method, this one does _not_ respect
            LIFO (last-in, first-out).
            This is due to the fact that attributes are stored in a random order when
            initializing the model.

            However, it will respect LIFO with respect to the internal `model_fields`.

        """
        if not self:
            raise KeyError(f"popitem(): {self.__class__.__name__} is empty")

        key = list(self.keys())[-1]
        value = self.pop(key)
        return key, value

    def reset_field(self, field: str) -> None:
        """Reset a field to its default value.

        Warning:
            This will remove/delete a field that is not part of the model schema.

        Parameters:
            field: The field to reset.

        """
        if field not in self:
            raise KeyError(f"Field {field!r} does not exist.")

        if field in self.model_fields:
            # Part of the model schema
            schema_field = True

            # Set the field to its default value
            setattr(self, field, self.model_fields[field].default)
        else:
            # Not part of the model schema, but part of the extras
            schema_field = False

            # Remove the field altogether
            if self.model_extra is None:
                raise RuntimeError("Model has no extra fields.")

            try:
                self.model_extra.pop(field)
            except KeyError as exc:
                raise RuntimeError(
                    f"Field {field!r} can not be found in the model fields or extras."
                ) from exc

        # Remove it from fields set by the user
        if field in self.model_fields_set:
            self.model_fields_set.remove(field)

        # Check the field has been properly reset
        if schema_field:
            if field not in self:
                raise RuntimeError(f"Field {field!r} was not reset as expected.")
            if self[field] != self.model_fields[field].default:
                raise RuntimeError(
                    f"Field {field!r} was not reset to its default value as expected."
                )
        else:
            if field in self:
                raise RuntimeError(f"Field {field!r} was not removed as expected.")
        if field in self.model_fields_set:
            raise RuntimeError(
                f"Field {field!r} was not removed from the set of user-set fields as "
                "expected."
            )

model_config = ConfigDict(extra='allow', validate_assignment=True, arbitrary_types_allowed=True) class-attribute instance-attribute

clear()

Source code in oteapi/models/genericconfig.py 
100
101
102
103
104
105
106
def clear(self) -> None:
    # Ignore the deprecation warning from `__delitem__`
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)

        for field in self.model_dump():
            del self[field]

get(key, default=None)

Source code in oteapi/models/genericconfig.py 
64
65
def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
    return getattr(self, key, default)

items()

Source code in oteapi/models/genericconfig.py 
55
56
def items(self):
    return self.model_dump().items()

keys()

Source code in oteapi/models/genericconfig.py 
58
59
def keys(self):
    return self.model_dump().keys()

pop(key, default=PydanticUndefined)

Source code in oteapi/models/genericconfig.py 
140
141
142
143
144
145
146
147
148
149
150
def pop(self, key: str, default: "Optional[Any]" = PydanticUndefined) -> "Any":
    value = self.get(key, default)
    if value == PydanticUndefined:
        raise KeyError(key)
    if key in self:
        # Ignore the deprecation warning from `__delitem__`
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)

            del self[key]
    return value

popitem()

MutableMapping popitem-method.

Important

Unlike the regular dict.popitem() method, this one does not respect LIFO (last-in, first-out). This is due to the fact that attributes are stored in a random order when initializing the model.

However, it will respect LIFO with respect to the internal model_fields.

Source code in oteapi/models/genericconfig.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def popitem(self) -> "Tuple[str, Any]":
    """MutableMapping `popitem`-method.

    Important:
        Unlike the regular `dict.popitem()` method, this one does _not_ respect
        LIFO (last-in, first-out).
        This is due to the fact that attributes are stored in a random order when
        initializing the model.

        However, it will respect LIFO with respect to the internal `model_fields`.

    """
    if not self:
        raise KeyError(f"popitem(): {self.__class__.__name__} is empty")

    key = list(self.keys())[-1]
    value = self.pop(key)
    return key, value

reset_field(field)

Reset a field to its default value.

Warning

This will remove/delete a field that is not part of the model schema.

Parameters:

Name Type Description Default
field str

The field to reset.

 required
Source code in oteapi/models/genericconfig.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def reset_field(self, field: str) -> None:
    """Reset a field to its default value.

    Warning:
        This will remove/delete a field that is not part of the model schema.

    Parameters:
        field: The field to reset.

    """
    if field not in self:
        raise KeyError(f"Field {field!r} does not exist.")

    if field in self.model_fields:
        # Part of the model schema
        schema_field = True

        # Set the field to its default value
        setattr(self, field, self.model_fields[field].default)
    else:
        # Not part of the model schema, but part of the extras
        schema_field = False

        # Remove the field altogether
        if self.model_extra is None:
            raise RuntimeError("Model has no extra fields.")

        try:
            self.model_extra.pop(field)
        except KeyError as exc:
            raise RuntimeError(
                f"Field {field!r} can not be found in the model fields or extras."
            ) from exc

    # Remove it from fields set by the user
    if field in self.model_fields_set:
        self.model_fields_set.remove(field)

    # Check the field has been properly reset
    if schema_field:
        if field not in self:
            raise RuntimeError(f"Field {field!r} was not reset as expected.")
        if self[field] != self.model_fields[field].default:
            raise RuntimeError(
                f"Field {field!r} was not reset to its default value as expected."
            )
    else:
        if field in self:
            raise RuntimeError(f"Field {field!r} was not removed as expected.")
    if field in self.model_fields_set:
        raise RuntimeError(
            f"Field {field!r} was not removed from the set of user-set fields as "
            "expected."
        )

update(other=None, **kwargs)

Source code in oteapi/models/genericconfig.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def update(  # type: ignore[override]
    self,
    other: "Optional[Union[Mapping[str, Any], Iterable[tuple[str, Any]]]]" = None,
    **kwargs,
) -> None:
    if other and isinstance(other, Mapping):
        for key, value in other.items():
            setattr(self, key, value)
    elif other and isinstance(other, BaseModel):
        for key, value in other:
            setattr(self, key, value)
    elif other and isinstance(other, Iterable):
        for entry in other:
            if not isinstance(entry, tuple):
                raise TypeError(
                    "`other` must be an iterable of tuples of length two."
                )
            if not len(entry) == 2:
                raise ValueError(
                    "`other` must be an iterable of objects of length two."
                )
        for key, value in other:  # type: ignore[misc]
            setattr(self, key, value)
    elif other:
        raise TypeError(
            "`other` must be of type `dict`, `Mapping`, `BaseModel` or "
            "`Iterable`, not `{type(other).__name__}`."
        )
    if kwargs:
        for key, value in kwargs.items():
            setattr(self, key, value)

values()

Source code in oteapi/models/genericconfig.py 
61
62
def values(self):
    return self.model_dump().values()

DataCacheConfig

Bases: AttrDict

DataCache Configuration.

This class should not be used directly as a configuration object for a strategy object, but only as a configuration field inside a configuration object.

Source code in oteapi/models/datacacheconfig.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class DataCacheConfig(AttrDict):
    """DataCache Configuration.

    This class should not be used directly as a configuration object
    for a strategy object, but only as a configuration field inside
    a configuration object.
    """

    cacheDir: Path = Field(Path("oteapi"), description="Cache directory.")
    accessKey: Optional[str] = Field(
        None,
        description="Key with which the downloaded content can be accessed. "
        "Should preferable be the hash (corresponding to `hashType`) of the "
        "content if it is known.",
    )
    hashType: str = Field(
        "md5",
        description="Hash algorithm to use for creating hash keys for stored "
        "data. Can be any algorithm supported by hashlib.",
    )
    expireTime: int = Field(
        3600 * 24 * 14,
        description="Number of seconds before the cache entry expires. "
        "Zero means no expiration. Default is two weeks.",
    )
    tag: Optional[str] = Field(
        None,
        description="Tag assigned to the downloaded content, typically "
        "identifying a session. Used with the `evict()` method to clean up a "
        "all cache entries with a given tag.",
    )

accessKey: Optional[str] = Field(None, description='Key with which the downloaded content can be accessed. Should preferable be the hash (corresponding to `hashType`) of the content if it is known.') class-attribute instance-attribute

cacheDir: Path = Field(Path('oteapi'), description='Cache directory.') class-attribute instance-attribute

expireTime: int = Field(3600 * 24 * 14, description='Number of seconds before the cache entry expires. Zero means no expiration. Default is two weeks.') class-attribute instance-attribute

hashType: str = Field('md5', description='Hash algorithm to use for creating hash keys for stored data. Can be any algorithm supported by hashlib.') class-attribute instance-attribute

tag: Optional[str] = Field(None, description='Tag assigned to the downloaded content, typically identifying a session. Used with the `evict()` method to clean up a all cache entries with a given tag.') class-attribute instance-attribute

FilterConfig

Bases: GenericConfig

Filter Strategy Data Configuration.

Source code in oteapi/models/filterconfig.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class FilterConfig(GenericConfig):
    """Filter Strategy Data Configuration."""

    filterType: str = Field(
        ..., description="Type of registered filter strategy. E.g., `filter/sql`."
    )
    query: Optional[str] = Field(None, description="Define a query operation.")
    condition: Optional[str] = Field(
        None,
        description="Logical statement indicating when a filter should be applied.",
    )
    limit: Optional[int] = Field(
        None, description="Number of items remaining after a filter expression."
    )

condition: Optional[str] = Field(None, description='Logical statement indicating when a filter should be applied.') class-attribute instance-attribute

filterType: str = Field(..., description='Type of registered filter strategy. E.g., `filter/sql`.') class-attribute instance-attribute

limit: Optional[int] = Field(None, description='Number of items remaining after a filter expression.') class-attribute instance-attribute

query: Optional[str] = Field(None, description='Define a query operation.') class-attribute instance-attribute

FunctionConfig

Bases: GenericConfig, SecretConfig

Function Strategy Data Configuration.

Source code in oteapi/models/functionconfig.py 
 9
10
11
12
13
14
15
class FunctionConfig(GenericConfig, SecretConfig):
    """Function Strategy Data Configuration."""

    functionType: str = Field(
        ...,
        description=("Type of registered function strategy."),
    )

functionType: str = Field(..., description='Type of registered function strategy.') class-attribute instance-attribute

GenericConfig

Bases: BaseModel

Generic class for configuration objects.

Special pydantic configuration settings:

  • validate_assignment Validate and cast set values. This is mainly relevant for sub-classes of AttrDict, where specific attributes have been defined.
  • arbitrary_types_allowed If a custom type is used for an attribute that doesn't have a validate() method, don't fail setting the attribute.
Source code in oteapi/models/genericconfig.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class GenericConfig(BaseModel):
    """Generic class for configuration objects.

    Special pydantic configuration settings:

    - **`validate_assignment`**
      Validate and cast set values.
      This is mainly relevant for sub-classes of `AttrDict`, where specific
      attributes have been defined.
    - **`arbitrary_types_allowed`**
      If a custom type is used for an attribute that doesn't have a `validate()`
      method, don't fail setting the attribute.

    """

    configuration: AttrDict = Field(
        AttrDict(),
        description="Model-specific configuration options which can either "
        "be given as key/value-pairs or set as attributes.",
    )

    description: str = Field(
        __doc__,
        description="A description of the configuration model.",
    )

    @classmethod
    def __init_subclass__(cls, **kwargs) -> None:
        """Initialize subclass descriptions with their docstrings."""
        cls.model_fields["description"].default = cls.__doc__

    model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True)

configuration: AttrDict = Field(AttrDict(), description='Model-specific configuration options which can either be given as key/value-pairs or set as attributes.') class-attribute instance-attribute

description: str = Field(__doc__, description='A description of the configuration model.') class-attribute instance-attribute

model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True) class-attribute instance-attribute

MappingConfig

Bases: GenericConfig

Mapping Strategy Data Configuration.

Source code in oteapi/models/mappingconfig.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class MappingConfig(GenericConfig):
    """Mapping Strategy Data Configuration."""

    mappingType: str = Field(
        ...,
        description="Type of registered mapping strategy.",
    )
    prefixes: Optional[Dict[str, str]] = Field(
        None,
        description=(
            "Dictionary of shortnames that expands to an IRI given as local "
            "value/IRI-expansion-pairs."
        ),
    )
    triples: Optional[Set[RDFTriple]] = Field(
        None,
        description="Set of RDF triples given as (subject, predicate, object).",
    )

mappingType: str = Field(..., description='Type of registered mapping strategy.') class-attribute instance-attribute

prefixes: Optional[Dict[str, str]] = Field(None, description='Dictionary of shortnames that expands to an IRI given as local value/IRI-expansion-pairs.') class-attribute instance-attribute

triples: Optional[Set[RDFTriple]] = Field(None, description='Set of RDF triples given as (subject, predicate, object).') class-attribute instance-attribute

ParserConfig

Bases: GenericConfig

Parser Strategy Data Configuration.

Source code in oteapi/models/parserconfig.py 
 8
 9
10
11
12
class ParserConfig(GenericConfig):
    """Parser Strategy Data Configuration."""

    parserType: str = Field(..., description="Type of registered parser strategy.")
    entity: AnyHttpUrl = Field(..., description="IRI to the entity or collection.")

entity: AnyHttpUrl = Field(..., description='IRI to the entity or collection.') class-attribute instance-attribute

parserType: str = Field(..., description='Type of registered parser strategy.') class-attribute instance-attribute

ResourceConfig

Bases: GenericConfig, SecretConfig

Resource Strategy Data Configuration.

Important

Either of the pairs of attributes downloadUrl/mediaType or accessUrl/accessService MUST be specified.

Source code in oteapi/models/resourceconfig.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class ResourceConfig(GenericConfig, SecretConfig):
    """Resource Strategy Data Configuration.

    Important:
        Either of the pairs of attributes `downloadUrl`/`mediaType` or
        `accessUrl`/`accessService` MUST be specified.

    """

    resourceType: Optional[str] = Field(
        None, description="Type of registered resource strategy."
    )

    downloadUrl: Optional[HostlessAnyUrl] = Field(
        None,
        description=(
            "Definition: The URL of the downloadable file in a given format. E.g. CSV "
            "file or RDF file.\n\nUsage: `downloadURL` *SHOULD* be used for the URL at"
            " which this distribution is available directly, typically through a HTTPS"
            " GET request or SFTP."
        ),
    )
    mediaType: Optional[str] = Field(
        None,
        description=(
            "The media type of the distribution as defined by IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]"
            ".\n\nUsage: This property *SHOULD* be used when the media"
            " type of the distribution is defined in IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]."
        ),
    )
    accessUrl: Optional[HostlessAnyUrl] = Field(
        None,
        description=(
            "A URL of the resource that gives access to a distribution of "
            "the dataset. E.g. landing page, feed, SPARQL endpoint.\n\nUsage: "
            "`accessURL` *SHOULD* be used for the URL of a service or location that "
            "can provide access to this distribution, typically through a Web form, "
            "query or API call.\n`downloadURL` is preferred for direct links to "
            "downloadable resources."
        ),
    )
    accessService: Optional[str] = Field(
        None,
        description=(
            "A data service that gives access to the distribution of the dataset."
        ),
    )
    license: Optional[str] = Field(
        None,
        description=(
            "A legal document under which the distribution is made available."
        ),
    )
    accessRights: Optional[str] = Field(
        None,
        description=(
            "A rights statement that concerns how the distribution is accessed."
        ),
    )
    publisher: Optional[str] = Field(
        None,
        description="The entity responsible for making the resource/item available.",
    )

    @model_validator(mode="after")
    def ensure_unique_url_pairs(self) -> "ResourceConfig":
        """Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

        It's fine to define them all, but at least one complete pair MUST be specified.
        """
        if not (
            all(getattr(self, _) for _ in ["downloadUrl", "mediaType"])
            or all(getattr(self, _) for _ in ["accessUrl", "accessService"])
        ):
            raise ValueError(
                "Either of the pairs of attributes downloadUrl/mediaType or "
                "accessUrl/accessService MUST be specified."
            )
        return self

accessRights: Optional[str] = Field(None, description='A rights statement that concerns how the distribution is accessed.') class-attribute instance-attribute

accessService: Optional[str] = Field(None, description='A data service that gives access to the distribution of the dataset.') class-attribute instance-attribute

accessUrl: Optional[HostlessAnyUrl] = Field(None, description='A URL of the resource that gives access to a distribution of the dataset. E.g. landing page, feed, SPARQL endpoint.\n\nUsage: `accessURL` *SHOULD* be used for the URL of a service or location that can provide access to this distribution, typically through a Web form, query or API call.\n`downloadURL` is preferred for direct links to downloadable resources.') class-attribute instance-attribute

downloadUrl: Optional[HostlessAnyUrl] = Field(None, description='Definition: The URL of the downloadable file in a given format. E.g. CSV file or RDF file.\n\nUsage: `downloadURL` *SHOULD* be used for the URL at which this distribution is available directly, typically through a HTTPS GET request or SFTP.') class-attribute instance-attribute

license: Optional[str] = Field(None, description='A legal document under which the distribution is made available.') class-attribute instance-attribute

mediaType: Optional[str] = Field(None, description='The media type of the distribution as defined by IANA [[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)].\n\nUsage: This property *SHOULD* be used when the media type of the distribution is defined in IANA [[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)].') class-attribute instance-attribute

publisher: Optional[str] = Field(None, description='The entity responsible for making the resource/item available.') class-attribute instance-attribute

resourceType: Optional[str] = Field(None, description='Type of registered resource strategy.') class-attribute instance-attribute

ensure_unique_url_pairs()

Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

It's fine to define them all, but at least one complete pair MUST be specified.

Source code in oteapi/models/resourceconfig.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@model_validator(mode="after")
def ensure_unique_url_pairs(self) -> "ResourceConfig":
    """Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

    It's fine to define them all, but at least one complete pair MUST be specified.
    """
    if not (
        all(getattr(self, _) for _ in ["downloadUrl", "mediaType"])
        or all(getattr(self, _) for _ in ["accessUrl", "accessService"])
    ):
        raise ValueError(
            "Either of the pairs of attributes downloadUrl/mediaType or "
            "accessUrl/accessService MUST be specified."
        )
    return self

SecretConfig

Bases: BaseModel

Simple model for handling secret in other config-models.

Source code in oteapi/models/secretconfig.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class SecretConfig(BaseModel):
    """Simple model for handling secret in other config-models."""

    user: Optional[TogglableSecretStr] = Field(
        None, description="User name for authentication."
    )
    password: Optional[TogglableSecretStr] = Field(
        None, description="Password for authentication."
    )
    token: Optional[TogglableSecretStr] = Field(
        None,
        description=(
            "An access token for providing access and meta data to an application."
        ),
    )
    client_id: Optional[TogglableSecretStr] = Field(
        None, description="Client ID for an OAUTH2 client."
    )
    client_secret: Optional[TogglableSecretStr] = Field(
        None, description="Client secret for an OAUTH2 client."
    )

client_id: Optional[TogglableSecretStr] = Field(None, description='Client ID for an OAUTH2 client.') class-attribute instance-attribute

client_secret: Optional[TogglableSecretStr] = Field(None, description='Client secret for an OAUTH2 client.') class-attribute instance-attribute

password: Optional[TogglableSecretStr] = Field(None, description='Password for authentication.') class-attribute instance-attribute

token: Optional[TogglableSecretStr] = Field(None, description='An access token for providing access and meta data to an application.') class-attribute instance-attribute

user: Optional[TogglableSecretStr] = Field(None, description='User name for authentication.') class-attribute instance-attribute

TransformationConfig

Bases: GenericConfig, SecretConfig

Transformation Strategy Data Configuration.

Source code in oteapi/models/transformationconfig.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class TransformationConfig(GenericConfig, SecretConfig):
    """Transformation Strategy Data Configuration."""

    transformationType: str = Field(
        ...,
        description=(
            "Type of registered transformation strategy. E.g., `celery/remote`."
        ),
    )
    name: Optional[str] = Field(
        None, description="Human-readable name of the transformation strategy."
    )
    due: Optional[datetime] = Field(
        None,
        description=(
            "Optional field to indicate a due data/time for when a transformation "
            "should finish."
        ),
    )
    priority: Optional[ProcessPriority] = Field(
        ProcessPriority.MEDIUM,
        description="Define the process priority of the transformation execution.",
    )

due: Optional[datetime] = Field(None, description='Optional field to indicate a due data/time for when a transformation should finish.') class-attribute instance-attribute

name: Optional[str] = Field(None, description='Human-readable name of the transformation strategy.') class-attribute instance-attribute

priority: Optional[ProcessPriority] = Field(ProcessPriority.MEDIUM, description='Define the process priority of the transformation execution.') class-attribute instance-attribute

transformationType: str = Field(..., description='Type of registered transformation strategy. E.g., `celery/remote`.') class-attribute instance-attribute

TransformationStatus

Bases: BaseModel

Return from transformation status.

Source code in oteapi/models/transformationconfig.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class TransformationStatus(BaseModel):
    """Return from transformation status."""

    id: str = Field(..., description="ID for the given transformation process.")
    status: Optional[str] = Field(
        None, description="Status for the transformation process."
    )
    messages: Optional[List[str]] = Field(
        None, description="Messages related to the transformation process."
    )
    created: Optional[datetime] = Field(
        None,
        description="Time of creation for the transformation process. Given in UTC.",
    )
    startTime: Optional[datetime] = Field(
        None, description="Time when the transformation process started. Given in UTC."
    )
    finishTime: Optional[datetime] = Field(
        None, description="Time when the tranformation process finished. Given in UTC."
    )

created: Optional[datetime] = Field(None, description='Time of creation for the transformation process. Given in UTC.') class-attribute instance-attribute

finishTime: Optional[datetime] = Field(None, description='Time when the tranformation process finished. Given in UTC.') class-attribute instance-attribute

id: str = Field(..., description='ID for the given transformation process.') class-attribute instance-attribute

messages: Optional[List[str]] = Field(None, description='Messages related to the transformation process.') class-attribute instance-attribute

startTime: Optional[datetime] = Field(None, description='Time when the transformation process started. Given in UTC.') class-attribute instance-attribute

status: Optional[str] = Field(None, description='Status for the transformation process.') class-attribute instance-attribute

TripleStoreConfig

Bases: GenericConfig, SecretConfig

TripleStore Configuration.

This is a configuration for the TripleStore.

This class should not be used directly as a configuration object for a strategy object, but only as a configuration field inside a configuration object.

Source code in oteapi/models/triplestoreconfig.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class TripleStoreConfig(GenericConfig, SecretConfig):
    """TripleStore Configuration.

    This is a configuration for the
    [`TripleStore`][oteapi.triplestore.triplestore.TripleStore].

    This class should not be used directly as a configuration object
    for a strategy object, but only as a configuration field inside
    a configuration object.
    """

    repositoryName: str = Field(
        ..., description="The repository name, where the mappings are stored."
    )
    agraphHost: str = Field(..., description="AllegroGraph host name.")
    agraphPort: int = Field(..., description="AllegroGraph port number.")

    # Exclude these inherited fields from serialization
    token: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields["token"]] = (
        SecretConfig.model_fields["token"].default
    )
    client_id: Annotated[
        ExcludeTogglableSecretStr, SecretConfig.model_fields["client_id"]
    ] = SecretConfig.model_fields["client_id"].default
    client_secret: Annotated[
        ExcludeTogglableSecretStr, SecretConfig.model_fields["client_secret"]
    ] = SecretConfig.model_fields["client_secret"].default

    @model_validator(mode="after")
    def ensure_user_pass(self) -> "TripleStoreConfig":
        """Ensure that user/password are set, since they are optional in the
        SecretConfig."""
        if not all(getattr(self, _) for _ in ["user", "password"]):
            raise ValueError("User and password must be defined.")
        return self

agraphHost: str = Field(..., description='AllegroGraph host name.') class-attribute instance-attribute

agraphPort: int = Field(..., description='AllegroGraph port number.') class-attribute instance-attribute

client_id: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[client_id]] = SecretConfig.model_fields['client_id'].default class-attribute instance-attribute

client_secret: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[client_secret]] = SecretConfig.model_fields['client_secret'].default class-attribute instance-attribute

repositoryName: str = Field(..., description='The repository name, where the mappings are stored.') class-attribute instance-attribute

token: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[token]] = SecretConfig.model_fields['token'].default class-attribute instance-attribute

ensure_user_pass()

Ensure that user/password are set, since they are optional in the SecretConfig.

Source code in oteapi/models/triplestoreconfig.py 
42
43
44
45
46
47
48
@model_validator(mode="after")
def ensure_user_pass(self) -> "TripleStoreConfig":
    """Ensure that user/password are set, since they are optional in the
    SecretConfig."""
    if not all(getattr(self, _) for _ in ["user", "password"]):
        raise ValueError("User and password must be defined.")
    return self

datacacheconfig

Pydantic DataCache Configuration Data Model.

DataCacheConfig

Bases: AttrDict

DataCache Configuration.

This class should not be used directly as a configuration object for a strategy object, but only as a configuration field inside a configuration object.

Source code in oteapi/models/datacacheconfig.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
class DataCacheConfig(AttrDict):
    """DataCache Configuration.

    This class should not be used directly as a configuration object
    for a strategy object, but only as a configuration field inside
    a configuration object.
    """

    cacheDir: Path = Field(Path("oteapi"), description="Cache directory.")
    accessKey: Optional[str] = Field(
        None,
        description="Key with which the downloaded content can be accessed. "
        "Should preferable be the hash (corresponding to `hashType`) of the "
        "content if it is known.",
    )
    hashType: str = Field(
        "md5",
        description="Hash algorithm to use for creating hash keys for stored "
        "data. Can be any algorithm supported by hashlib.",
    )
    expireTime: int = Field(
        3600 * 24 * 14,
        description="Number of seconds before the cache entry expires. "
        "Zero means no expiration. Default is two weeks.",
    )
    tag: Optional[str] = Field(
        None,
        description="Tag assigned to the downloaded content, typically "
        "identifying a session. Used with the `evict()` method to clean up a "
        "all cache entries with a given tag.",
    )

accessKey: Optional[str] = Field(None, description='Key with which the downloaded content can be accessed. Should preferable be the hash (corresponding to `hashType`) of the content if it is known.') class-attribute instance-attribute

cacheDir: Path = Field(Path('oteapi'), description='Cache directory.') class-attribute instance-attribute

expireTime: int = Field(3600 * 24 * 14, description='Number of seconds before the cache entry expires. Zero means no expiration. Default is two weeks.') class-attribute instance-attribute

hashType: str = Field('md5', description='Hash algorithm to use for creating hash keys for stored data. Can be any algorithm supported by hashlib.') class-attribute instance-attribute

tag: Optional[str] = Field(None, description='Tag assigned to the downloaded content, typically identifying a session. Used with the `evict()` method to clean up a all cache entries with a given tag.') class-attribute instance-attribute

filterconfig

Pydantic Filter Configuration Data Model.

FilterConfig

Bases: GenericConfig

Filter Strategy Data Configuration.

Source code in oteapi/models/filterconfig.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
class FilterConfig(GenericConfig):
    """Filter Strategy Data Configuration."""

    filterType: str = Field(
        ..., description="Type of registered filter strategy. E.g., `filter/sql`."
    )
    query: Optional[str] = Field(None, description="Define a query operation.")
    condition: Optional[str] = Field(
        None,
        description="Logical statement indicating when a filter should be applied.",
    )
    limit: Optional[int] = Field(
        None, description="Number of items remaining after a filter expression."
    )

condition: Optional[str] = Field(None, description='Logical statement indicating when a filter should be applied.') class-attribute instance-attribute

filterType: str = Field(..., description='Type of registered filter strategy. E.g., `filter/sql`.') class-attribute instance-attribute

limit: Optional[int] = Field(None, description='Number of items remaining after a filter expression.') class-attribute instance-attribute

query: Optional[str] = Field(None, description='Define a query operation.') class-attribute instance-attribute

functionconfig

Pydantic Function Configuration Data Model.

FunctionConfig

Bases: GenericConfig, SecretConfig

Function Strategy Data Configuration.

Source code in oteapi/models/functionconfig.py 
 9
10
11
12
13
14
15
class FunctionConfig(GenericConfig, SecretConfig):
    """Function Strategy Data Configuration."""

    functionType: str = Field(
        ...,
        description=("Type of registered function strategy."),
    )

functionType: str = Field(..., description='Type of registered function strategy.') class-attribute instance-attribute

genericconfig

Generic data model for configuration attributes.

AttrDict

Bases: BaseModel, MutableMapping

An object whose attributes can also be accessed through subscription, like with a dictionary.

Special pydantic configuration settings:

  • extra Allow any attributes/fields to be defined - this is what makes this pydantic model an attribute dictionary.
  • validate_assignment Validate and cast set values. This is mainly relevant for sub-classes of AttrDict, where specific attributes have been defined.
  • arbitrary_types_allowed If a custom type is used for an attribute that doesn't have a validate() method, don't fail setting the attribute.
Source code in oteapi/models/genericconfig.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class AttrDict(BaseModel, MutableMapping):
    """An object whose attributes can also be accessed through
    subscription, like with a dictionary.

    Special pydantic configuration settings:

    - **`extra`**
      Allow any attributes/fields to be defined - this is what makes this pydantic
      model an attribute dictionary.
    - **`validate_assignment`**
      Validate and cast set values.
      This is mainly relevant for sub-classes of `AttrDict`, where specific
      attributes have been defined.
    - **`arbitrary_types_allowed`**
      If a custom type is used for an attribute that doesn't have a `validate()`
      method, don't fail setting the attribute.

    """

    model_config = ConfigDict(
        extra="allow", validate_assignment=True, arbitrary_types_allowed=True
    )

    # Collection methods
    def __contains__(self, key: object) -> bool:
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        return hasattr(self, key)

    def __len__(self) -> int:
        return len(self.model_dump())

    # Mapping methods
    def __getitem__(self, key: "Any") -> "Any":
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            return getattr(self, key)
        except AttributeError as exc:
            raise KeyError(key) from exc

    def items(self):
        return self.model_dump().items()

    def keys(self):
        return self.model_dump().keys()

    def values(self):
        return self.model_dump().values()

    def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
        return getattr(self, key, default)

    def __eq__(self, value: object) -> bool:
        if isinstance(value, Mapping):
            return self.model_dump() == value
        if isinstance(value, BaseModel):
            return BaseModel.__eq__(self, value)
        return False

    # MutableMapping methods
    def __setitem__(self, key: "Any", value: "Any") -> None:
        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            return self.__setattr__(key, value)
        except AttributeError as exc:
            raise KeyError(key) from exc

    def __delitem__(self, key: "Any") -> None:
        warnings.warn(
            "Item deletion used to reset fields to their default values. To keep using"
            " this functionality, use the `reset_field()` method.",
            DeprecationWarning,
        )

        if not isinstance(key, str):
            raise TypeError(f"Keys must be of type `str`, not `{type(key).__name__}`.")
        try:
            self.__delattr__(key)
        except AttributeError as exc:
            raise KeyError(key) from exc
        else:
            if key in self.model_fields_set:
                self.model_fields_set.remove(key)

    def clear(self) -> None:
        # Ignore the deprecation warning from `__delitem__`
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)

            for field in self.model_dump():
                del self[field]

    def update(  # type: ignore[override]
        self,
        other: "Optional[Union[Mapping[str, Any], Iterable[tuple[str, Any]]]]" = None,
        **kwargs,
    ) -> None:
        if other and isinstance(other, Mapping):
            for key, value in other.items():
                setattr(self, key, value)
        elif other and isinstance(other, BaseModel):
            for key, value in other:
                setattr(self, key, value)
        elif other and isinstance(other, Iterable):
            for entry in other:
                if not isinstance(entry, tuple):
                    raise TypeError(
                        "`other` must be an iterable of tuples of length two."
                    )
                if not len(entry) == 2:
                    raise ValueError(
                        "`other` must be an iterable of objects of length two."
                    )
            for key, value in other:  # type: ignore[misc]
                setattr(self, key, value)
        elif other:
            raise TypeError(
                "`other` must be of type `dict`, `Mapping`, `BaseModel` or "
                "`Iterable`, not `{type(other).__name__}`."
            )
        if kwargs:
            for key, value in kwargs.items():
                setattr(self, key, value)

    def pop(self, key: str, default: "Optional[Any]" = PydanticUndefined) -> "Any":
        value = self.get(key, default)
        if value == PydanticUndefined:
            raise KeyError(key)
        if key in self:
            # Ignore the deprecation warning from `__delitem__`
            with warnings.catch_warnings():
                warnings.simplefilter("ignore", DeprecationWarning)

                del self[key]
        return value

    def popitem(self) -> "Tuple[str, Any]":
        """MutableMapping `popitem`-method.

        Important:
            Unlike the regular `dict.popitem()` method, this one does _not_ respect
            LIFO (last-in, first-out).
            This is due to the fact that attributes are stored in a random order when
            initializing the model.

            However, it will respect LIFO with respect to the internal `model_fields`.

        """
        if not self:
            raise KeyError(f"popitem(): {self.__class__.__name__} is empty")

        key = list(self.keys())[-1]
        value = self.pop(key)
        return key, value

    def reset_field(self, field: str) -> None:
        """Reset a field to its default value.

        Warning:
            This will remove/delete a field that is not part of the model schema.

        Parameters:
            field: The field to reset.

        """
        if field not in self:
            raise KeyError(f"Field {field!r} does not exist.")

        if field in self.model_fields:
            # Part of the model schema
            schema_field = True

            # Set the field to its default value
            setattr(self, field, self.model_fields[field].default)
        else:
            # Not part of the model schema, but part of the extras
            schema_field = False

            # Remove the field altogether
            if self.model_extra is None:
                raise RuntimeError("Model has no extra fields.")

            try:
                self.model_extra.pop(field)
            except KeyError as exc:
                raise RuntimeError(
                    f"Field {field!r} can not be found in the model fields or extras."
                ) from exc

        # Remove it from fields set by the user
        if field in self.model_fields_set:
            self.model_fields_set.remove(field)

        # Check the field has been properly reset
        if schema_field:
            if field not in self:
                raise RuntimeError(f"Field {field!r} was not reset as expected.")
            if self[field] != self.model_fields[field].default:
                raise RuntimeError(
                    f"Field {field!r} was not reset to its default value as expected."
                )
        else:
            if field in self:
                raise RuntimeError(f"Field {field!r} was not removed as expected.")
        if field in self.model_fields_set:
            raise RuntimeError(
                f"Field {field!r} was not removed from the set of user-set fields as "
                "expected."
            )

model_config = ConfigDict(extra='allow', validate_assignment=True, arbitrary_types_allowed=True) class-attribute instance-attribute

clear()

Source code in oteapi/models/genericconfig.py 
100
101
102
103
104
105
106
def clear(self) -> None:
    # Ignore the deprecation warning from `__delitem__`
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", DeprecationWarning)

        for field in self.model_dump():
            del self[field]

get(key, default=None)

Source code in oteapi/models/genericconfig.py 
64
65
def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
    return getattr(self, key, default)

items()

Source code in oteapi/models/genericconfig.py 
55
56
def items(self):
    return self.model_dump().items()

keys()

Source code in oteapi/models/genericconfig.py 
58
59
def keys(self):
    return self.model_dump().keys()

pop(key, default=PydanticUndefined)

Source code in oteapi/models/genericconfig.py 
140
141
142
143
144
145
146
147
148
149
150
def pop(self, key: str, default: "Optional[Any]" = PydanticUndefined) -> "Any":
    value = self.get(key, default)
    if value == PydanticUndefined:
        raise KeyError(key)
    if key in self:
        # Ignore the deprecation warning from `__delitem__`
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", DeprecationWarning)

            del self[key]
    return value

popitem()

MutableMapping popitem-method.

Important

Unlike the regular dict.popitem() method, this one does not respect LIFO (last-in, first-out). This is due to the fact that attributes are stored in a random order when initializing the model.

However, it will respect LIFO with respect to the internal model_fields.

Source code in oteapi/models/genericconfig.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def popitem(self) -> "Tuple[str, Any]":
    """MutableMapping `popitem`-method.

    Important:
        Unlike the regular `dict.popitem()` method, this one does _not_ respect
        LIFO (last-in, first-out).
        This is due to the fact that attributes are stored in a random order when
        initializing the model.

        However, it will respect LIFO with respect to the internal `model_fields`.

    """
    if not self:
        raise KeyError(f"popitem(): {self.__class__.__name__} is empty")

    key = list(self.keys())[-1]
    value = self.pop(key)
    return key, value

reset_field(field)

Reset a field to its default value.

Warning

This will remove/delete a field that is not part of the model schema.

Parameters:

Name Type Description Default
field str

The field to reset.

 required
Source code in oteapi/models/genericconfig.py 
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def reset_field(self, field: str) -> None:
    """Reset a field to its default value.

    Warning:
        This will remove/delete a field that is not part of the model schema.

    Parameters:
        field: The field to reset.

    """
    if field not in self:
        raise KeyError(f"Field {field!r} does not exist.")

    if field in self.model_fields:
        # Part of the model schema
        schema_field = True

        # Set the field to its default value
        setattr(self, field, self.model_fields[field].default)
    else:
        # Not part of the model schema, but part of the extras
        schema_field = False

        # Remove the field altogether
        if self.model_extra is None:
            raise RuntimeError("Model has no extra fields.")

        try:
            self.model_extra.pop(field)
        except KeyError as exc:
            raise RuntimeError(
                f"Field {field!r} can not be found in the model fields or extras."
            ) from exc

    # Remove it from fields set by the user
    if field in self.model_fields_set:
        self.model_fields_set.remove(field)

    # Check the field has been properly reset
    if schema_field:
        if field not in self:
            raise RuntimeError(f"Field {field!r} was not reset as expected.")
        if self[field] != self.model_fields[field].default:
            raise RuntimeError(
                f"Field {field!r} was not reset to its default value as expected."
            )
    else:
        if field in self:
            raise RuntimeError(f"Field {field!r} was not removed as expected.")
    if field in self.model_fields_set:
        raise RuntimeError(
            f"Field {field!r} was not removed from the set of user-set fields as "
            "expected."
        )

update(other=None, **kwargs)

Source code in oteapi/models/genericconfig.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def update(  # type: ignore[override]
    self,
    other: "Optional[Union[Mapping[str, Any], Iterable[tuple[str, Any]]]]" = None,
    **kwargs,
) -> None:
    if other and isinstance(other, Mapping):
        for key, value in other.items():
            setattr(self, key, value)
    elif other and isinstance(other, BaseModel):
        for key, value in other:
            setattr(self, key, value)
    elif other and isinstance(other, Iterable):
        for entry in other:
            if not isinstance(entry, tuple):
                raise TypeError(
                    "`other` must be an iterable of tuples of length two."
                )
            if not len(entry) == 2:
                raise ValueError(
                    "`other` must be an iterable of objects of length two."
                )
        for key, value in other:  # type: ignore[misc]
            setattr(self, key, value)
    elif other:
        raise TypeError(
            "`other` must be of type `dict`, `Mapping`, `BaseModel` or "
            "`Iterable`, not `{type(other).__name__}`."
        )
    if kwargs:
        for key, value in kwargs.items():
            setattr(self, key, value)

values()

Source code in oteapi/models/genericconfig.py 
61
62
def values(self):
    return self.model_dump().values()

GenericConfig

Bases: BaseModel

Generic class for configuration objects.

Special pydantic configuration settings:

  • validate_assignment Validate and cast set values. This is mainly relevant for sub-classes of AttrDict, where specific attributes have been defined.
  • arbitrary_types_allowed If a custom type is used for an attribute that doesn't have a validate() method, don't fail setting the attribute.
Source code in oteapi/models/genericconfig.py 
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
class GenericConfig(BaseModel):
    """Generic class for configuration objects.

    Special pydantic configuration settings:

    - **`validate_assignment`**
      Validate and cast set values.
      This is mainly relevant for sub-classes of `AttrDict`, where specific
      attributes have been defined.
    - **`arbitrary_types_allowed`**
      If a custom type is used for an attribute that doesn't have a `validate()`
      method, don't fail setting the attribute.

    """

    configuration: AttrDict = Field(
        AttrDict(),
        description="Model-specific configuration options which can either "
        "be given as key/value-pairs or set as attributes.",
    )

    description: str = Field(
        __doc__,
        description="A description of the configuration model.",
    )

    @classmethod
    def __init_subclass__(cls, **kwargs) -> None:
        """Initialize subclass descriptions with their docstrings."""
        cls.model_fields["description"].default = cls.__doc__

    model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True)

configuration: AttrDict = Field(AttrDict(), description='Model-specific configuration options which can either be given as key/value-pairs or set as attributes.') class-attribute instance-attribute

description: str = Field(__doc__, description='A description of the configuration model.') class-attribute instance-attribute

model_config = ConfigDict(validate_assignment=True, arbitrary_types_allowed=True) class-attribute instance-attribute

mappingconfig

Pydantic Mapping Configuration Data Model.

RDFTriple = Tuple[str, str, str] module-attribute

MappingConfig

Bases: GenericConfig

Mapping Strategy Data Configuration.

Source code in oteapi/models/mappingconfig.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
class MappingConfig(GenericConfig):
    """Mapping Strategy Data Configuration."""

    mappingType: str = Field(
        ...,
        description="Type of registered mapping strategy.",
    )
    prefixes: Optional[Dict[str, str]] = Field(
        None,
        description=(
            "Dictionary of shortnames that expands to an IRI given as local "
            "value/IRI-expansion-pairs."
        ),
    )
    triples: Optional[Set[RDFTriple]] = Field(
        None,
        description="Set of RDF triples given as (subject, predicate, object).",
    )

mappingType: str = Field(..., description='Type of registered mapping strategy.') class-attribute instance-attribute

prefixes: Optional[Dict[str, str]] = Field(None, description='Dictionary of shortnames that expands to an IRI given as local value/IRI-expansion-pairs.') class-attribute instance-attribute

triples: Optional[Set[RDFTriple]] = Field(None, description='Set of RDF triples given as (subject, predicate, object).') class-attribute instance-attribute

parserconfig

Pydantic Parser Configuration Data Model.

ParserConfig

Bases: GenericConfig

Parser Strategy Data Configuration.

Source code in oteapi/models/parserconfig.py 
 8
 9
10
11
12
class ParserConfig(GenericConfig):
    """Parser Strategy Data Configuration."""

    parserType: str = Field(..., description="Type of registered parser strategy.")
    entity: AnyHttpUrl = Field(..., description="IRI to the entity or collection.")

entity: AnyHttpUrl = Field(..., description='IRI to the entity or collection.') class-attribute instance-attribute

parserType: str = Field(..., description='Type of registered parser strategy.') class-attribute instance-attribute

resourceconfig

Pydantic Resource Configuration Data Model.

HostlessAnyUrl = Annotated[Url, UrlConstraints(host_required=False)] module-attribute

ResourceConfig

Bases: GenericConfig, SecretConfig

Resource Strategy Data Configuration.

Important

Either of the pairs of attributes downloadUrl/mediaType or accessUrl/accessService MUST be specified.

Source code in oteapi/models/resourceconfig.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class ResourceConfig(GenericConfig, SecretConfig):
    """Resource Strategy Data Configuration.

    Important:
        Either of the pairs of attributes `downloadUrl`/`mediaType` or
        `accessUrl`/`accessService` MUST be specified.

    """

    resourceType: Optional[str] = Field(
        None, description="Type of registered resource strategy."
    )

    downloadUrl: Optional[HostlessAnyUrl] = Field(
        None,
        description=(
            "Definition: The URL of the downloadable file in a given format. E.g. CSV "
            "file or RDF file.\n\nUsage: `downloadURL` *SHOULD* be used for the URL at"
            " which this distribution is available directly, typically through a HTTPS"
            " GET request or SFTP."
        ),
    )
    mediaType: Optional[str] = Field(
        None,
        description=(
            "The media type of the distribution as defined by IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]"
            ".\n\nUsage: This property *SHOULD* be used when the media"
            " type of the distribution is defined in IANA "
            "[[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)]."
        ),
    )
    accessUrl: Optional[HostlessAnyUrl] = Field(
        None,
        description=(
            "A URL of the resource that gives access to a distribution of "
            "the dataset. E.g. landing page, feed, SPARQL endpoint.\n\nUsage: "
            "`accessURL` *SHOULD* be used for the URL of a service or location that "
            "can provide access to this distribution, typically through a Web form, "
            "query or API call.\n`downloadURL` is preferred for direct links to "
            "downloadable resources."
        ),
    )
    accessService: Optional[str] = Field(
        None,
        description=(
            "A data service that gives access to the distribution of the dataset."
        ),
    )
    license: Optional[str] = Field(
        None,
        description=(
            "A legal document under which the distribution is made available."
        ),
    )
    accessRights: Optional[str] = Field(
        None,
        description=(
            "A rights statement that concerns how the distribution is accessed."
        ),
    )
    publisher: Optional[str] = Field(
        None,
        description="The entity responsible for making the resource/item available.",
    )

    @model_validator(mode="after")
    def ensure_unique_url_pairs(self) -> "ResourceConfig":
        """Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

        It's fine to define them all, but at least one complete pair MUST be specified.
        """
        if not (
            all(getattr(self, _) for _ in ["downloadUrl", "mediaType"])
            or all(getattr(self, _) for _ in ["accessUrl", "accessService"])
        ):
            raise ValueError(
                "Either of the pairs of attributes downloadUrl/mediaType or "
                "accessUrl/accessService MUST be specified."
            )
        return self

accessRights: Optional[str] = Field(None, description='A rights statement that concerns how the distribution is accessed.') class-attribute instance-attribute

accessService: Optional[str] = Field(None, description='A data service that gives access to the distribution of the dataset.') class-attribute instance-attribute

accessUrl: Optional[HostlessAnyUrl] = Field(None, description='A URL of the resource that gives access to a distribution of the dataset. E.g. landing page, feed, SPARQL endpoint.\n\nUsage: `accessURL` *SHOULD* be used for the URL of a service or location that can provide access to this distribution, typically through a Web form, query or API call.\n`downloadURL` is preferred for direct links to downloadable resources.') class-attribute instance-attribute

downloadUrl: Optional[HostlessAnyUrl] = Field(None, description='Definition: The URL of the downloadable file in a given format. E.g. CSV file or RDF file.\n\nUsage: `downloadURL` *SHOULD* be used for the URL at which this distribution is available directly, typically through a HTTPS GET request or SFTP.') class-attribute instance-attribute

license: Optional[str] = Field(None, description='A legal document under which the distribution is made available.') class-attribute instance-attribute

mediaType: Optional[str] = Field(None, description='The media type of the distribution as defined by IANA [[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)].\n\nUsage: This property *SHOULD* be used when the media type of the distribution is defined in IANA [[IANA-MEDIA-TYPES](https://www.w3.org/TR/vocab-dcat-2/#bib-iana-media-types)].') class-attribute instance-attribute

publisher: Optional[str] = Field(None, description='The entity responsible for making the resource/item available.') class-attribute instance-attribute

resourceType: Optional[str] = Field(None, description='Type of registered resource strategy.') class-attribute instance-attribute

ensure_unique_url_pairs()

Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

It's fine to define them all, but at least one complete pair MUST be specified.

Source code in oteapi/models/resourceconfig.py 
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
@model_validator(mode="after")
def ensure_unique_url_pairs(self) -> "ResourceConfig":
    """Ensure either downloadUrl/mediaType or accessUrl/accessService are defined.

    It's fine to define them all, but at least one complete pair MUST be specified.
    """
    if not (
        all(getattr(self, _) for _ in ["downloadUrl", "mediaType"])
        or all(getattr(self, _) for _ in ["accessUrl", "accessService"])
    ):
        raise ValueError(
            "Either of the pairs of attributes downloadUrl/mediaType or "
            "accessUrl/accessService MUST be specified."
        )
    return self

secretconfig

AttrDict for specifying user credentials or secrets.

TogglableSecretStr = Annotated[SecretStr, PlainSerializer(lambda value: value.get_secret_value() if settings.expose_secrets else str(value), return_type=str, when_used='json-unless-none')] module-attribute

Annotated type alias for a secret string that can be toggled to be exposed or not.

SecretConfig

Bases: BaseModel

Simple model for handling secret in other config-models.

Source code in oteapi/models/secretconfig.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
class SecretConfig(BaseModel):
    """Simple model for handling secret in other config-models."""

    user: Optional[TogglableSecretStr] = Field(
        None, description="User name for authentication."
    )
    password: Optional[TogglableSecretStr] = Field(
        None, description="Password for authentication."
    )
    token: Optional[TogglableSecretStr] = Field(
        None,
        description=(
            "An access token for providing access and meta data to an application."
        ),
    )
    client_id: Optional[TogglableSecretStr] = Field(
        None, description="Client ID for an OAUTH2 client."
    )
    client_secret: Optional[TogglableSecretStr] = Field(
        None, description="Client secret for an OAUTH2 client."
    )

client_id: Optional[TogglableSecretStr] = Field(None, description='Client ID for an OAUTH2 client.') class-attribute instance-attribute

client_secret: Optional[TogglableSecretStr] = Field(None, description='Client secret for an OAUTH2 client.') class-attribute instance-attribute

password: Optional[TogglableSecretStr] = Field(None, description='Password for authentication.') class-attribute instance-attribute

token: Optional[TogglableSecretStr] = Field(None, description='An access token for providing access and meta data to an application.') class-attribute instance-attribute

user: Optional[TogglableSecretStr] = Field(None, description='User name for authentication.') class-attribute instance-attribute

transformationconfig

Pydantic Transformation Configuration Data Model.

A transformation status data model is provided as well. This data model represents what should be returned from the strategy's status() method.

ProcessPriority

Bases: str, Enum

Defining process priority enumerators.

Process priorities:

  • Low
  • Medium
  • High
Source code in oteapi/models/transformationconfig.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
class ProcessPriority(str, Enum):
    """Defining process priority enumerators.

    Process priorities:

    - Low
    - Medium
    - High

    """

    LOW = "Low"
    MEDIUM = "Medium"
    HIGH = "High"

HIGH = 'High' class-attribute instance-attribute

LOW = 'Low' class-attribute instance-attribute

MEDIUM = 'Medium' class-attribute instance-attribute

TransformationConfig

Bases: GenericConfig, SecretConfig

Transformation Strategy Data Configuration.

Source code in oteapi/models/transformationconfig.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
class TransformationConfig(GenericConfig, SecretConfig):
    """Transformation Strategy Data Configuration."""

    transformationType: str = Field(
        ...,
        description=(
            "Type of registered transformation strategy. E.g., `celery/remote`."
        ),
    )
    name: Optional[str] = Field(
        None, description="Human-readable name of the transformation strategy."
    )
    due: Optional[datetime] = Field(
        None,
        description=(
            "Optional field to indicate a due data/time for when a transformation "
            "should finish."
        ),
    )
    priority: Optional[ProcessPriority] = Field(
        ProcessPriority.MEDIUM,
        description="Define the process priority of the transformation execution.",
    )

due: Optional[datetime] = Field(None, description='Optional field to indicate a due data/time for when a transformation should finish.') class-attribute instance-attribute

name: Optional[str] = Field(None, description='Human-readable name of the transformation strategy.') class-attribute instance-attribute

priority: Optional[ProcessPriority] = Field(ProcessPriority.MEDIUM, description='Define the process priority of the transformation execution.') class-attribute instance-attribute

transformationType: str = Field(..., description='Type of registered transformation strategy. E.g., `celery/remote`.') class-attribute instance-attribute

TransformationStatus

Bases: BaseModel

Return from transformation status.

Source code in oteapi/models/transformationconfig.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
class TransformationStatus(BaseModel):
    """Return from transformation status."""

    id: str = Field(..., description="ID for the given transformation process.")
    status: Optional[str] = Field(
        None, description="Status for the transformation process."
    )
    messages: Optional[List[str]] = Field(
        None, description="Messages related to the transformation process."
    )
    created: Optional[datetime] = Field(
        None,
        description="Time of creation for the transformation process. Given in UTC.",
    )
    startTime: Optional[datetime] = Field(
        None, description="Time when the transformation process started. Given in UTC."
    )
    finishTime: Optional[datetime] = Field(
        None, description="Time when the tranformation process finished. Given in UTC."
    )

created: Optional[datetime] = Field(None, description='Time of creation for the transformation process. Given in UTC.') class-attribute instance-attribute

finishTime: Optional[datetime] = Field(None, description='Time when the tranformation process finished. Given in UTC.') class-attribute instance-attribute

id: str = Field(..., description='ID for the given transformation process.') class-attribute instance-attribute

messages: Optional[List[str]] = Field(None, description='Messages related to the transformation process.') class-attribute instance-attribute

startTime: Optional[datetime] = Field(None, description='Time when the transformation process started. Given in UTC.') class-attribute instance-attribute

status: Optional[str] = Field(None, description='Status for the transformation process.') class-attribute instance-attribute

triplestoreconfig

Pydantic TripleStore Configuration Data Model.

ExcludeTogglableSecretStr = Annotated[Optional[TogglableSecretStr], Field(exclude=True)] module-attribute

Annotated type alias for excluding a togglable secret from serialization.

TripleStoreConfig

Bases: GenericConfig, SecretConfig

TripleStore Configuration.

This is a configuration for the TripleStore.

This class should not be used directly as a configuration object for a strategy object, but only as a configuration field inside a configuration object.

Source code in oteapi/models/triplestoreconfig.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class TripleStoreConfig(GenericConfig, SecretConfig):
    """TripleStore Configuration.

    This is a configuration for the
    [`TripleStore`][oteapi.triplestore.triplestore.TripleStore].

    This class should not be used directly as a configuration object
    for a strategy object, but only as a configuration field inside
    a configuration object.
    """

    repositoryName: str = Field(
        ..., description="The repository name, where the mappings are stored."
    )
    agraphHost: str = Field(..., description="AllegroGraph host name.")
    agraphPort: int = Field(..., description="AllegroGraph port number.")

    # Exclude these inherited fields from serialization
    token: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields["token"]] = (
        SecretConfig.model_fields["token"].default
    )
    client_id: Annotated[
        ExcludeTogglableSecretStr, SecretConfig.model_fields["client_id"]
    ] = SecretConfig.model_fields["client_id"].default
    client_secret: Annotated[
        ExcludeTogglableSecretStr, SecretConfig.model_fields["client_secret"]
    ] = SecretConfig.model_fields["client_secret"].default

    @model_validator(mode="after")
    def ensure_user_pass(self) -> "TripleStoreConfig":
        """Ensure that user/password are set, since they are optional in the
        SecretConfig."""
        if not all(getattr(self, _) for _ in ["user", "password"]):
            raise ValueError("User and password must be defined.")
        return self

agraphHost: str = Field(..., description='AllegroGraph host name.') class-attribute instance-attribute

agraphPort: int = Field(..., description='AllegroGraph port number.') class-attribute instance-attribute

client_id: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[client_id]] = SecretConfig.model_fields['client_id'].default class-attribute instance-attribute

client_secret: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[client_secret]] = SecretConfig.model_fields['client_secret'].default class-attribute instance-attribute

repositoryName: str = Field(..., description='The repository name, where the mappings are stored.') class-attribute instance-attribute

token: Annotated[ExcludeTogglableSecretStr, SecretConfig.model_fields[token]] = SecretConfig.model_fields['token'].default class-attribute instance-attribute

ensure_user_pass()

Ensure that user/password are set, since they are optional in the SecretConfig.

Source code in oteapi/models/triplestoreconfig.py 
42
43
44
45
46
47
48
@model_validator(mode="after")
def ensure_user_pass(self) -> "TripleStoreConfig":
    """Ensure that user/password are set, since they are optional in the
    SecretConfig."""
    if not all(getattr(self, _) for _ in ["user", "password"]):
        raise ValueError("User and password must be defined.")
    return self