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.

AttrDict

Bases: BaseModel, Mapping

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

Source code in oteapi/models/genericconfig.py 
10
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
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
class AttrDict(BaseModel, Mapping):
    """An object whose attributes can also be accessed through
    subscription, like with a dictionary."""

    class Config:
        """Class for configuration of pydantic models."""

        extra = "allow"

    def __contains__(self, name: "Any") -> bool:
        """Enable using the 'in' operator on this object."""
        return self.__dict__.__contains__(name)

    def __delitem__(self, key: str) -> None:
        """Enable deletion access through subscription."""
        if key in dir(self):
            self.__delattr__(key)
            if key in self.__fields__:
                del self.__fields__[key]
                self.__fields_set__.remove(key)  # pylint: disable=no-member
        else:
            raise KeyError(key)

    def __getitem__(self, key: str) -> "Any":
        """Enable read access through subscription."""
        if key in dir(self):
            return getattr(self, key)
        raise KeyError(key)

    def __setattr__(self, name: str, value: "Any") -> None:
        """Extend BaseModel.__setattr__ with type-checking."""
        if name in self.__dict__ and self.__dict__[name]:
            target_type = type(self.__dict__[name])
            if not isinstance(value, target_type):
                raise TypeError(
                    "Mapped value must be subclass of " + target_type.__name__
                )
        super().__setattr__(name, value)

    def __setitem__(self, key: str, value: "Any") -> None:
        """Enable write access through subscription."""
        self.__setattr__(key, value)

    def __len__(self):
        """Return number of items."""
        return self.__dict__.__len__()

    def __iter__(self):
        """Enable **unpacking."""
        return self.__dict__.__iter__()

    def items(self):
        """Return a view of all (key, value) pairs."""
        return self.__dict__.items()

    def keys(self):
        """Return a view of all keys."""
        return self.__dict__.keys()

    def values(self):
        """Return a view of all values."""
        return self.__dict__.values()

    def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
        """Mapping `get`-method."""
        return self.__dict__.get(key, default)

    def __ne__(self, other: "Any") -> bool:
        if isinstance(other, BaseModel):
            return self.dict() != other.dict()
        return self.dict() != other

    def update(
        self, other: "Optional[Union[dict, AttrDict, Iterable[Any]]]" = None, **kwargs
    ) -> None:
        """MutableMapping `update`-method."""
        if other and isinstance(other, (dict, self.__class__)):
            for key, value in other.items():
                setattr(self, key, value)
        elif other and isinstance(other, Iterable):
            for entry in other:
                if not len(entry) == 2:
                    raise ValueError(
                        "`other` must be an iterable of objects of length two."
                    )
            for key, value in other:
                setattr(self, key, value)
        if kwargs:
            for key, value in kwargs.items():
                setattr(self, key, value)

Config

Class for configuration of pydantic models.

Source code in oteapi/models/genericconfig.py 
14
15
16
17
class Config:
    """Class for configuration of pydantic models."""

    extra = "allow"

__contains__(self, name)

Enable using the 'in' operator on this object.

Source code in oteapi/models/genericconfig.py 
19
20
21
def __contains__(self, name: "Any") -> bool:
    """Enable using the 'in' operator on this object."""
    return self.__dict__.__contains__(name)

__delitem__(self, key)

Enable deletion access through subscription.

Source code in oteapi/models/genericconfig.py 
23
24
25
26
27
28
29
30
31
def __delitem__(self, key: str) -> None:
    """Enable deletion access through subscription."""
    if key in dir(self):
        self.__delattr__(key)
        if key in self.__fields__:
            del self.__fields__[key]
            self.__fields_set__.remove(key)  # pylint: disable=no-member
    else:
        raise KeyError(key)

__getitem__(self, key)

Enable read access through subscription.

Source code in oteapi/models/genericconfig.py 
33
34
35
36
37
def __getitem__(self, key: str) -> "Any":
    """Enable read access through subscription."""
    if key in dir(self):
        return getattr(self, key)
    raise KeyError(key)

__iter__(self)

Enable **unpacking.

Source code in oteapi/models/genericconfig.py 
57
58
59
def __iter__(self):
    """Enable **unpacking."""
    return self.__dict__.__iter__()

__len__(self)

Return number of items.

Source code in oteapi/models/genericconfig.py 
53
54
55
def __len__(self):
    """Return number of items."""
    return self.__dict__.__len__()

__setattr__(self, name, value)

Extend BaseModel.setattr with type-checking.

Source code in oteapi/models/genericconfig.py 
39
40
41
42
43
44
45
46
47
def __setattr__(self, name: str, value: "Any") -> None:
    """Extend BaseModel.__setattr__ with type-checking."""
    if name in self.__dict__ and self.__dict__[name]:
        target_type = type(self.__dict__[name])
        if not isinstance(value, target_type):
            raise TypeError(
                "Mapped value must be subclass of " + target_type.__name__
            )
    super().__setattr__(name, value)

__setitem__(self, key, value)

Enable write access through subscription.

Source code in oteapi/models/genericconfig.py 
49
50
51
def __setitem__(self, key: str, value: "Any") -> None:
    """Enable write access through subscription."""
    self.__setattr__(key, value)

get(self, key, default=None)

Mapping get-method.

Source code in oteapi/models/genericconfig.py 
73
74
75
def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
    """Mapping `get`-method."""
    return self.__dict__.get(key, default)

items(self)

Return a view of all (key, value) pairs.

Source code in oteapi/models/genericconfig.py 
61
62
63
def items(self):
    """Return a view of all (key, value) pairs."""
    return self.__dict__.items()

keys(self)

Return a view of all keys.

Source code in oteapi/models/genericconfig.py 
65
66
67
def keys(self):
    """Return a view of all keys."""
    return self.__dict__.keys()

update(self, other=None, **kwargs)

MutableMapping update-method.

Source code in oteapi/models/genericconfig.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def update(
    self, other: "Optional[Union[dict, AttrDict, Iterable[Any]]]" = None, **kwargs
) -> None:
    """MutableMapping `update`-method."""
    if other and isinstance(other, (dict, self.__class__)):
        for key, value in other.items():
            setattr(self, key, value)
    elif other and isinstance(other, Iterable):
        for entry in other:
            if not len(entry) == 2:
                raise ValueError(
                    "`other` must be an iterable of objects of length two."
                )
        for key, value in other:
            setattr(self, key, value)
    if kwargs:
        for key, value in kwargs.items():
            setattr(self, key, value)

values(self)

Return a view of all values.

Source code in oteapi/models/genericconfig.py 
69
70
71
def values(self):
    """Return a view of all values."""
    return self.__dict__.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 
 9
10
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
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("oteapi", description="Cache directory.")
    accessKey: 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: 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.",
    )

FilterConfig

Bases: GenericConfig

Filter Strategy Data Configuration.

Source code in oteapi/models/filterconfig.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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."
    )

FunctionConfig

Bases: GenericConfig

Function Strategy Data Configuration.

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

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

GenericConfig

Bases: BaseModel

Generic class for configuration objects.

Source code in oteapi/models/genericconfig.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class GenericConfig(BaseModel):
    """Generic class for configuration objects."""

    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) -> None:
        """Initialize subclass descriptions with their docstrings."""
        cls.__fields__["description"].default = cls.__doc__

__init_subclass__(cls)

Initialize subclass descriptions with their docstrings.

Source code in oteapi/models/genericconfig.py 
116
117
118
119
@classmethod
def __init_subclass__(cls) -> None:
    """Initialize subclass descriptions with their docstrings."""
    cls.__fields__["description"].default = cls.__doc__

MappingConfig

Bases: GenericConfig

Mapping Strategy Data Configuration.

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

    mappingType: str = Field(
        ..., description="Type of registered mapping strategy. E.g., `mapping/demo`."
    )
    prefixes: Optional[Dict[str, str]] = Field(
        None,
        description=(
            "List of shortnames that expands to an IRI "
            "given as local value/IRI-expansion-pairs."
        ),
    )
    triples: Optional[List[SemanticTriple]] = Field(  # type: ignore[valid-type]
        None,
        description="List of semantic triples given as (subject, predicate, object).",
    )

ResourceConfig

Bases: GenericConfig

Important

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

Resource Strategy Data Configuration.

Source code in oteapi/models/resourceconfig.py 
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
class ResourceConfig(GenericConfig):
    """Resource Strategy Data Configuration.

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

    """

    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.",
    )

    @root_validator
    def ensure_unique_url_pairs(  # pylint: disable=no-self-use
        cls, values: Dict[str, Any]
    ) -> Dict[str, Any]:
        """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(values.get(_) for _ in ["downloadUrl", "mediaType"])
            or all(values.get(_) for _ in ["accessUrl", "accessService"])
        ):
            raise ValueError(
                "Either of the pairs of attributes downloadUrl/mediaType or "
                "accessUrl/accessService MUST be specified."
            )
        return values

ensure_unique_url_pairs(cls, values)

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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@root_validator
def ensure_unique_url_pairs(  # pylint: disable=no-self-use
    cls, values: Dict[str, Any]
) -> Dict[str, Any]:
    """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(values.get(_) for _ in ["downloadUrl", "mediaType"])
        or all(values.get(_) for _ in ["accessUrl", "accessService"])
    ):
        raise ValueError(
            "Either of the pairs of attributes downloadUrl/mediaType or "
            "accessUrl/accessService MUST be specified."
        )
    return values

SessionUpdate

Bases: AttrDict

Session Update Data Model for returning values.

Source code in oteapi/models/sessionupdate.py 
5
6
class SessionUpdate(AttrDict):
    """Session Update Data Model for returning values."""

TransformationConfig

Bases: GenericConfig

Transformation Strategy Data Configuration.

Source code in oteapi/models/transformationconfig.py 
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
class TransformationConfig(GenericConfig):
    """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[PriorityEnum] = Field(
        PriorityEnum.MEDIUM,
        description="Define the process priority of the transformation execution.",
    )
    secret: Optional[str] = Field(
        None,
        description="Authorization secret given when running a transformation.",
    )

TransformationStatus

Bases: BaseModel

Return from transformation status.

Source code in oteapi/models/transformationconfig.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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."
    )

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 
 9
10
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
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("oteapi", description="Cache directory.")
    accessKey: 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: 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.",
    )

filterconfig

Pydantic Filter Configuration Data Model.

FilterConfig

Bases: GenericConfig

Filter Strategy Data Configuration.

Source code in oteapi/models/filterconfig.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
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."
    )

functionconfig

Pydantic Function Configuration Data Model.

FunctionConfig

Bases: GenericConfig

Function Strategy Data Configuration.

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

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

genericconfig

Generic data model for configuration attributes.

AttrDict

Bases: BaseModel, Mapping

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

Source code in oteapi/models/genericconfig.py 
10
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
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
class AttrDict(BaseModel, Mapping):
    """An object whose attributes can also be accessed through
    subscription, like with a dictionary."""

    class Config:
        """Class for configuration of pydantic models."""

        extra = "allow"

    def __contains__(self, name: "Any") -> bool:
        """Enable using the 'in' operator on this object."""
        return self.__dict__.__contains__(name)

    def __delitem__(self, key: str) -> None:
        """Enable deletion access through subscription."""
        if key in dir(self):
            self.__delattr__(key)
            if key in self.__fields__:
                del self.__fields__[key]
                self.__fields_set__.remove(key)  # pylint: disable=no-member
        else:
            raise KeyError(key)

    def __getitem__(self, key: str) -> "Any":
        """Enable read access through subscription."""
        if key in dir(self):
            return getattr(self, key)
        raise KeyError(key)

    def __setattr__(self, name: str, value: "Any") -> None:
        """Extend BaseModel.__setattr__ with type-checking."""
        if name in self.__dict__ and self.__dict__[name]:
            target_type = type(self.__dict__[name])
            if not isinstance(value, target_type):
                raise TypeError(
                    "Mapped value must be subclass of " + target_type.__name__
                )
        super().__setattr__(name, value)

    def __setitem__(self, key: str, value: "Any") -> None:
        """Enable write access through subscription."""
        self.__setattr__(key, value)

    def __len__(self):
        """Return number of items."""
        return self.__dict__.__len__()

    def __iter__(self):
        """Enable **unpacking."""
        return self.__dict__.__iter__()

    def items(self):
        """Return a view of all (key, value) pairs."""
        return self.__dict__.items()

    def keys(self):
        """Return a view of all keys."""
        return self.__dict__.keys()

    def values(self):
        """Return a view of all values."""
        return self.__dict__.values()

    def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
        """Mapping `get`-method."""
        return self.__dict__.get(key, default)

    def __ne__(self, other: "Any") -> bool:
        if isinstance(other, BaseModel):
            return self.dict() != other.dict()
        return self.dict() != other

    def update(
        self, other: "Optional[Union[dict, AttrDict, Iterable[Any]]]" = None, **kwargs
    ) -> None:
        """MutableMapping `update`-method."""
        if other and isinstance(other, (dict, self.__class__)):
            for key, value in other.items():
                setattr(self, key, value)
        elif other and isinstance(other, Iterable):
            for entry in other:
                if not len(entry) == 2:
                    raise ValueError(
                        "`other` must be an iterable of objects of length two."
                    )
            for key, value in other:
                setattr(self, key, value)
        if kwargs:
            for key, value in kwargs.items():
                setattr(self, key, value)

Config

Class for configuration of pydantic models.

Source code in oteapi/models/genericconfig.py 
14
15
16
17
class Config:
    """Class for configuration of pydantic models."""

    extra = "allow"

__contains__(self, name)

Enable using the 'in' operator on this object.

Source code in oteapi/models/genericconfig.py 
19
20
21
def __contains__(self, name: "Any") -> bool:
    """Enable using the 'in' operator on this object."""
    return self.__dict__.__contains__(name)

__delitem__(self, key)

Enable deletion access through subscription.

Source code in oteapi/models/genericconfig.py 
23
24
25
26
27
28
29
30
31
def __delitem__(self, key: str) -> None:
    """Enable deletion access through subscription."""
    if key in dir(self):
        self.__delattr__(key)
        if key in self.__fields__:
            del self.__fields__[key]
            self.__fields_set__.remove(key)  # pylint: disable=no-member
    else:
        raise KeyError(key)

__getitem__(self, key)

Enable read access through subscription.

Source code in oteapi/models/genericconfig.py 
33
34
35
36
37
def __getitem__(self, key: str) -> "Any":
    """Enable read access through subscription."""
    if key in dir(self):
        return getattr(self, key)
    raise KeyError(key)

__iter__(self)

Enable **unpacking.

Source code in oteapi/models/genericconfig.py 
57
58
59
def __iter__(self):
    """Enable **unpacking."""
    return self.__dict__.__iter__()

__len__(self)

Return number of items.

Source code in oteapi/models/genericconfig.py 
53
54
55
def __len__(self):
    """Return number of items."""
    return self.__dict__.__len__()

__setattr__(self, name, value)

Extend BaseModel.setattr with type-checking.

Source code in oteapi/models/genericconfig.py 
39
40
41
42
43
44
45
46
47
def __setattr__(self, name: str, value: "Any") -> None:
    """Extend BaseModel.__setattr__ with type-checking."""
    if name in self.__dict__ and self.__dict__[name]:
        target_type = type(self.__dict__[name])
        if not isinstance(value, target_type):
            raise TypeError(
                "Mapped value must be subclass of " + target_type.__name__
            )
    super().__setattr__(name, value)

__setitem__(self, key, value)

Enable write access through subscription.

Source code in oteapi/models/genericconfig.py 
49
50
51
def __setitem__(self, key: str, value: "Any") -> None:
    """Enable write access through subscription."""
    self.__setattr__(key, value)

get(self, key, default=None)

Mapping get-method.

Source code in oteapi/models/genericconfig.py 
73
74
75
def get(self, key: str, default: "Optional[Any]" = None) -> "Any":
    """Mapping `get`-method."""
    return self.__dict__.get(key, default)

items(self)

Return a view of all (key, value) pairs.

Source code in oteapi/models/genericconfig.py 
61
62
63
def items(self):
    """Return a view of all (key, value) pairs."""
    return self.__dict__.items()

keys(self)

Return a view of all keys.

Source code in oteapi/models/genericconfig.py 
65
66
67
def keys(self):
    """Return a view of all keys."""
    return self.__dict__.keys()

update(self, other=None, **kwargs)

MutableMapping update-method.

Source code in oteapi/models/genericconfig.py 
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def update(
    self, other: "Optional[Union[dict, AttrDict, Iterable[Any]]]" = None, **kwargs
) -> None:
    """MutableMapping `update`-method."""
    if other and isinstance(other, (dict, self.__class__)):
        for key, value in other.items():
            setattr(self, key, value)
    elif other and isinstance(other, Iterable):
        for entry in other:
            if not len(entry) == 2:
                raise ValueError(
                    "`other` must be an iterable of objects of length two."
                )
        for key, value in other:
            setattr(self, key, value)
    if kwargs:
        for key, value in kwargs.items():
            setattr(self, key, value)

values(self)

Return a view of all values.

Source code in oteapi/models/genericconfig.py 
69
70
71
def values(self):
    """Return a view of all values."""
    return self.__dict__.values()

GenericConfig

Bases: BaseModel

Generic class for configuration objects.

Source code in oteapi/models/genericconfig.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
class GenericConfig(BaseModel):
    """Generic class for configuration objects."""

    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) -> None:
        """Initialize subclass descriptions with their docstrings."""
        cls.__fields__["description"].default = cls.__doc__

__init_subclass__(cls)

Initialize subclass descriptions with their docstrings.

Source code in oteapi/models/genericconfig.py 
116
117
118
119
@classmethod
def __init_subclass__(cls) -> None:
    """Initialize subclass descriptions with their docstrings."""
    cls.__fields__["description"].default = cls.__doc__

mappingconfig

Pydantic Mapping Configuration Data Model.

MappingConfig

Bases: GenericConfig

Mapping Strategy Data Configuration.

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

    mappingType: str = Field(
        ..., description="Type of registered mapping strategy. E.g., `mapping/demo`."
    )
    prefixes: Optional[Dict[str, str]] = Field(
        None,
        description=(
            "List of shortnames that expands to an IRI "
            "given as local value/IRI-expansion-pairs."
        ),
    )
    triples: Optional[List[SemanticTriple]] = Field(  # type: ignore[valid-type]
        None,
        description="List of semantic triples given as (subject, predicate, object).",
    )

resourceconfig

Pydantic Resource Configuration Data Model.

HostlessAnyUrl

Bases: AnyUrl

AnyUrl, but allow not having a host.

Source code in oteapi/models/resourceconfig.py 
 9
10
11
12
class HostlessAnyUrl(AnyUrl):
    """AnyUrl, but allow not having a host."""

    host_required = False

ResourceConfig

Bases: GenericConfig

Important

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

Resource Strategy Data Configuration.

Source code in oteapi/models/resourceconfig.py 
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
class ResourceConfig(GenericConfig):
    """Resource Strategy Data Configuration.

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

    """

    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.",
    )

    @root_validator
    def ensure_unique_url_pairs(  # pylint: disable=no-self-use
        cls, values: Dict[str, Any]
    ) -> Dict[str, Any]:
        """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(values.get(_) for _ in ["downloadUrl", "mediaType"])
            or all(values.get(_) for _ in ["accessUrl", "accessService"])
        ):
            raise ValueError(
                "Either of the pairs of attributes downloadUrl/mediaType or "
                "accessUrl/accessService MUST be specified."
            )
        return values

ensure_unique_url_pairs(cls, values)

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 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
@root_validator
def ensure_unique_url_pairs(  # pylint: disable=no-self-use
    cls, values: Dict[str, Any]
) -> Dict[str, Any]:
    """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(values.get(_) for _ in ["downloadUrl", "mediaType"])
        or all(values.get(_) for _ in ["accessUrl", "accessService"])
    ):
        raise ValueError(
            "Either of the pairs of attributes downloadUrl/mediaType or "
            "accessUrl/accessService MUST be specified."
        )
    return values

sessionupdate

Pydantic SessionUpdate Data Model.

SessionUpdate

Bases: AttrDict

Session Update Data Model for returning values.

Source code in oteapi/models/sessionupdate.py 
5
6
class SessionUpdate(AttrDict):
    """Session Update Data Model for returning values."""

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.

PriorityEnum

Bases: str, Enum

Defining process priority enumerators.

Process priorities:

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

    Process priorities:

    - Low
    - Medium
    - High

    """

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

TransformationConfig

Bases: GenericConfig

Transformation Strategy Data Configuration.

Source code in oteapi/models/transformationconfig.py 
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
class TransformationConfig(GenericConfig):
    """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[PriorityEnum] = Field(
        PriorityEnum.MEDIUM,
        description="Define the process priority of the transformation execution.",
    )
    secret: Optional[str] = Field(
        None,
        description="Authorization secret given when running a transformation.",
    )

TransformationStatus

Bases: BaseModel

Return from transformation status.

Source code in oteapi/models/transformationconfig.py 
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
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."
    )
Back to top