Skip to content

mapping

Mapping filter strategy.

DLiteMappingConfig

Bases: MappingConfig

DLite mapping strategy config.

Source code in oteapi_dlite/strategies/mapping.py 
33
34
35
36
37
38
39
40
41
class DLiteMappingConfig(MappingConfig):
    """DLite mapping strategy config."""

    configuration: Annotated[
        DLiteMappingStrategyConfig,
        Field(
            description="DLite mapping strategy-specific configuration.",
        ),
    ] = DLiteMappingStrategyConfig()

configuration: Annotated[DLiteMappingStrategyConfig, Field(description='DLite mapping strategy-specific configuration.')] = DLiteMappingStrategyConfig() class-attribute instance-attribute

DLiteMappingStrategy

Strategy for a mapping.

Registers strategies:

  • ("mappingType", "mappings")
Source code in oteapi_dlite/strategies/mapping.py 
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
@dataclass
class DLiteMappingStrategy:
    """Strategy for a mapping.

    **Registers strategies**:

    - `("mappingType", "mappings")`

    """

    mapping_config: DLiteMappingConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize strategy."""
        if session is None:
            session = {}

        ts = get_triplestore(session)

        if self.mapping_config.prefixes:
            for prefix, iri in self.mapping_config.prefixes.items():
                ts.bind(prefix, iri)

        if self.mapping_config.triples:
            ts.add_triples(
                [
                    [
                        ts.expand_iri(t) if isinstance(t, str) else t
                        for t in triple
                    ]
                    for triple in self.mapping_config.triples
                ]
            )

        coll = get_collection(session)
        update_collection(coll)
        return DLiteSessionUpdate(collection_id=coll.uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute strategy and return a dictionary."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

mapping_config: DLiteMappingConfig instance-attribute

get(session=None)

Execute strategy and return a dictionary.

Source code in oteapi_dlite/strategies/mapping.py 
84
85
86
87
88
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute strategy and return a dictionary."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

initialize(session=None)

Initialize strategy.

Source code in oteapi_dlite/strategies/mapping.py 
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
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize strategy."""
    if session is None:
        session = {}

    ts = get_triplestore(session)

    if self.mapping_config.prefixes:
        for prefix, iri in self.mapping_config.prefixes.items():
            ts.bind(prefix, iri)

    if self.mapping_config.triples:
        ts.add_triples(
            [
                [
                    ts.expand_iri(t) if isinstance(t, str) else t
                    for t in triple
                ]
                for triple in self.mapping_config.triples
            ]
        )

    coll = get_collection(session)
    update_collection(coll)
    return DLiteSessionUpdate(collection_id=coll.uuid)

DLiteMappingStrategyConfig

Bases: AttrDict

Configuration for a DLite mapping filter.

Source code in oteapi_dlite/strategies/mapping.py 
22
23
24
25
26
27
28
29
30
class DLiteMappingStrategyConfig(AttrDict):
    """Configuration for a DLite mapping filter."""

    datamodel: Annotated[
        Optional[AnyUrl],
        Field(
            description="URI of the datamodel that is mapped.",
        ),
    ] = None

datamodel: Annotated[Optional[AnyUrl], Field(description='URI of the datamodel that is mapped.')] = None class-attribute instance-attribute