Skip to content

serialise

Filter for serialisation using DLite.

SerialiseConfig

Bases: AttrDict

DLite serialise-specific configurations.

Source code in oteapi_dlite/strategies/serialise.py 
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
class SerialiseConfig(AttrDict):
    """DLite serialise-specific configurations."""

    driver: Annotated[
        str,
        Field(
            description="Name of DLite plugin used for serialisation.",
        ),
    ]
    location: Annotated[
        Path,
        Field(
            description="Path or URL to serialise to.",
        ),
    ]
    options: Annotated[
        Optional[str],
        Field(
            description="Options passed to the driver.",
        ),
    ] = ""
    labels: Annotated[
        Optional[Sequence[str]],
        Field(
            None,
            description=(
                "Optional sequence of labels in the collection to serialise.  "
                "The default is to serialise the entire collection."
            ),
        ),
    ] = None

driver: Annotated[str, Field(description='Name of DLite plugin used for serialisation.')] instance-attribute

labels: Annotated[Optional[Sequence[str]], Field(None, description='Optional sequence of labels in the collection to serialise. The default is to serialise the entire collection.')] = None class-attribute instance-attribute

location: Annotated[Path, Field(description='Path or URL to serialise to.')] instance-attribute

options: Annotated[Optional[str], Field(description='Options passed to the driver.')] = '' class-attribute instance-attribute

SerialiseFilterConfig

Bases: FilterConfig

Filter config for serialise.

Source code in oteapi_dlite/strategies/serialise.py 
54
55
56
57
58
59
60
61
62
class SerialiseFilterConfig(FilterConfig):
    """Filter config for serialise."""

    configuration: Annotated[
        SerialiseConfig,
        Field(
            description="Serialise-specific configurations.",
        ),
    ]

configuration: Annotated[SerialiseConfig, Field(description='Serialise-specific configurations.')] instance-attribute

SerialiseStrategy

Filter for serialisation using DLite.

Registers strategies:

  • ("filterType", "dlite_serialise")
Source code in oteapi_dlite/strategies/serialise.py 
 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
@dataclass
class SerialiseStrategy:
    """Filter for serialisation using DLite.

    **Registers strategies**:

    - `("filterType", "dlite_serialise")`

    """

    filter_config: SerialiseFilterConfig

    def initialize(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Initialize."""
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)

    def get(
        self, session: Optional[dict[str, Any]] = None
    ) -> DLiteSessionUpdate:
        """Execute the strategy."""
        config = self.filter_config.configuration

        coll = get_collection(session)

        storage = dlite.Storage(
            driver_or_url=config.driver,
            location=str(config.location),
            options=config.options,
        )
        if config.labels is None:
            coll.save_to_storage(storage)
        else:
            for label in config.labels:
                inst = coll.get(label)
                inst.save_to_storage(storage)

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

filter_config: SerialiseFilterConfig instance-attribute

get(session=None)

Execute the strategy.

Source code in oteapi_dlite/strategies/serialise.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def get(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Execute the strategy."""
    config = self.filter_config.configuration

    coll = get_collection(session)

    storage = dlite.Storage(
        driver_or_url=config.driver,
        location=str(config.location),
        options=config.options,
    )
    if config.labels is None:
        coll.save_to_storage(storage)
    else:
        for label in config.labels:
            inst = coll.get(label)
            inst.save_to_storage(storage)

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

initialize(session=None)

Initialize.

Source code in oteapi_dlite/strategies/serialise.py 
77
78
79
80
81
def initialize(
    self, session: Optional[dict[str, Any]] = None
) -> DLiteSessionUpdate:
    """Initialize."""
    return DLiteSessionUpdate(collection_id=get_collection(session).uuid)