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.

datacacheconfig

Pydantic DataCache Configuration Data Model.

DataCacheConfig (BaseModel) pydantic-model

DataCache Configuration.

Source code in oteapi/models/datacacheconfig.py 
class DataCacheConfig(BaseModel):
    """DataCache Configuration."""

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

accessKey: str pydantic-field

Key with which the downloaded content can be accessed. Should preferable be the hash (corresponding to hashType) of the content if it is known.

cacheDir: Path pydantic-field

Cache directory.

expireTime: int pydantic-field

Number of seconds before the cache entry expires. Zero means no expiration. Default is two weeks.

hashType: str pydantic-field

Hash algorithm to use for creating hash keys for stored data. Can be any algorithm supported by hashlib.

tag: str pydantic-field

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 (BaseModel) pydantic-model

Filter Strategy Data Configuration.

Source code in oteapi/models/filterconfig.py 
class FilterConfig(BaseModel):
    """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."
    )
    configuration: Optional[Dict] = Field(
        None,
        description="Filter-specific configuration options given as key/value-pairs.",
    )

condition: str pydantic-field

Logical statement indicating when a filter should be applied.

configuration: Dict pydantic-field

Filter-specific configuration options given as key/value-pairs.

filterType: str pydantic-field required

Type of registered filter strategy. E.g., filter/sql.

limit: int pydantic-field

Number of items remaining after a filter expression.

query: str pydantic-field

Define a query operation.

mappingconfig

Pydantic Mapping Configuration Data Model.

MappingConfig (BaseModel) pydantic-model

Mapping Strategy Data Configuration.

Source code in oteapi/models/mappingconfig.py 
class MappingConfig(BaseModel):
    """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).",
    )
    configuration: Optional[Dict] = Field(
        None,
        description="Mapping-specific configuration options given as key/value-pairs.",
    )

configuration: Dict pydantic-field

Mapping-specific configuration options given as key/value-pairs.

mappingType: str pydantic-field required

Type of registered mapping strategy. E.g., mapping/demo.

prefixes: Dict[str, str] pydantic-field

List of shortnames that expands to an IRI given as local value/IRI-expansion-pairs.

triples: List[types.ConstrainedListValue] pydantic-field

List of semantic triples given as (subject, predicate, object).

resourceconfig

Pydantic Resource Configuration Data Model.

ResourceConfig (BaseModel) pydantic-model

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 
class ResourceConfig(BaseModel):
    """Resource Strategy Data Configuration.

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

    """

    downloadUrl: Optional[AnyUrl] = 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[AnyUrl] = 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."
        ),
    )
    description: Optional[str] = Field(
        None, description="A free-text account of the distribution."
    )
    publisher: Optional[str] = Field(
        None,
        description="The entity responsible for making the resource/item available.",
    )
    configuration: Union[DataCacheConfig, Dict[str, Any]] = Field(
        {},
        description="Resource-specific configuration options given as key/value-pairs.",
    )

    @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

accessRights: str pydantic-field

A rights statement that concerns how the distribution is accessed.

accessService: str pydantic-field

A data service that gives access to the distribution of the dataset.

accessUrl: AnyUrl pydantic-field

A URL of the resource that gives access to a distribution of the dataset. E.g. landing page, feed, SPARQL endpoint.

Usage: 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. downloadURL is preferred for direct links to downloadable resources.

configuration: Union[oteapi.models.datacacheconfig.DataCacheConfig, Dict[str, Any]] pydantic-field

Resource-specific configuration options given as key/value-pairs.

description: str pydantic-field

A free-text account of the distribution.

downloadUrl: AnyUrl pydantic-field

Definition: The URL of the downloadable file in a given format. E.g. CSV file or RDF file.

Usage: downloadURL SHOULD be used for the URL at which this distribution is available directly, typically through a HTTPS GET request or SFTP.

license: str pydantic-field

A legal document under which the distribution is made available.

mediaType: str pydantic-field

The media type of the distribution as defined by IANA [IANA-MEDIA-TYPES].

Usage: This property SHOULD be used when the media type of the distribution is defined in IANA [IANA-MEDIA-TYPES].

publisher: str pydantic-field

The entity responsible for making the resource/item available.

ensure_unique_url_pairs(values) classmethod

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 
@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

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 (str, Enum)

Defining process priority enumerators.

Process priorities:

  • Low
  • Medium
  • High
Source code in oteapi/models/transformationconfig.py 
class PriorityEnum(str, Enum):
    """Defining process priority enumerators.

    Process priorities:

    - Low
    - Medium
    - High

    """

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

TransformationConfig (BaseModel) pydantic-model

Transformation Strategy Data Configuration.

Source code in oteapi/models/transformationconfig.py 
class TransformationConfig(BaseModel):
    """Transformation Strategy Data Configuration."""

    transformation_type: 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."
    )
    description: Optional[str] = Field(
        None, description="A free-text account of the transformation."
    )
    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.",
    )
    configuration: Optional[Dict] = Field(
        None,
        description=(
            "Transformation-specific configuration options given as key/value-pairs."
        ),
    )

configuration: Dict pydantic-field

Transformation-specific configuration options given as key/value-pairs.

description: str pydantic-field

A free-text account of the transformation.

due: datetime pydantic-field

Optional field to indicate a due data/time for when a transformation should finish.

name: str pydantic-field

Human-readable name of the transformation strategy.

priority: PriorityEnum pydantic-field

Define the process priority of the transformation execution.

secret: str pydantic-field

Authorization secret given when running a transformation.

transformation_type: str pydantic-field required

Type of registered transformation strategy. E.g., celery/remote.

TransformationStatus (BaseModel) pydantic-model

Return from transformation status.

Source code in oteapi/models/transformationconfig.py 
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: datetime pydantic-field

Time of creation for the transformation process. Given in UTC.

finishTime: datetime pydantic-field

Time when the tranformation process finished. Given in UTC.

id: str pydantic-field required

ID for the given transformation process.

messages: List[str] pydantic-field

Messages related to the transformation process.

startTime: datetime pydantic-field

Time when the transformation process started. Given in UTC.

status: str pydantic-field

Status for the transformation process.

Back to top