Skip to content

parse

Demo strategy class for text/json.

OPTIMADEParseStrategy

Parse strategy for JSON.

Implements strategies:

  • ("parserType", "parser/OPTIMADE")
Source code in oteapi_optimade/strategies/parse.py 
 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
@dataclass
class OPTIMADEParseStrategy:
    """Parse strategy for JSON.

    **Implements strategies**:

    - `("parserType", "parser/OPTIMADE")`

    """

    parse_config: OPTIMADEParseConfig

    def initialize(self) -> AttrDict:
        """Initialize strategy.

        This method will be called through the `/initialize` endpoint of the OTE-API
        Services.

        Returns:
            An update model of key/value-pairs to be stored in the session-specific
            context from services.

        """
        return AttrDict()

    def get(self) -> OPTIMADEParseResult:
        """Request and parse an OPTIMADE response using OPT.

        This method will be called through the strategy-specific endpoint of the
        OTE-API Services.

        Configuration values provided in `resource_config.configuration` take
        precedence over the derived values from `downloadUrl`.

        Workflow:

        1. Request OPTIMADE response.
        2. Parse as an OPTIMADE Python tools (OPT) pydantic response model.

        Returns:
            An update model of key/value-pairs to be stored in the session-specific
            context from services.

        """
        if (
            self.parse_config.configuration.downloadUrl is None
            or self.parse_config.configuration.mediaType is None
        ):
            raise OPTIMADEParseError(
                "Missing downloadUrl or mediaType in configuration."
            )

        cache = DataCache(self.parse_config.configuration.datacache_config)
        if self.parse_config.configuration.downloadUrl in cache:
            response: dict[str, Any] = cache.get(
                self.parse_config.configuration.downloadUrl
            )
        elif (
            self.parse_config.configuration.datacache_config.accessKey
            and self.parse_config.configuration.datacache_config.accessKey in cache
        ):
            response = cache.get(
                self.parse_config.configuration.datacache_config.accessKey
            )
        else:
            download_config = self.parse_config.configuration.model_copy()
            download_output = create_strategy("download", download_config).get()
            response = {"json": json.loads(cache.get(download_output.pop("key")))}

        if (
            not response.get("ok", True)
            or (
                response.get("status_code", 200) < 200
                or response.get("status_code", 200) >= 300
            )
            or "errors" in response.get("json", {})
        ):
            # Error response
            try:
                response_object = ErrorResponse(**response.get("json", {}))
            except ValidationError as exc:
                error_message = "Could not validate an error response."
                LOGGER.error(
                    "%s\nValidationError: " "%s\nresponse=%r",
                    error_message,
                    exc,
                    response,
                )
                raise OPTIMADEParseError(error_message) from exc
        else:
            # Successful response
            response_model = (
                self.parse_config.configuration.downloadUrl.response_model()
            )
            LOGGER.debug("response_model=%r", response_model)
            if response_model:
                if not isinstance(response_model, tuple):
                    response_model = (response_model,)

                for model_cls in response_model:
                    try:
                        response_object = model_cls(**response.get("json", {}))
                    except ValidationError:
                        pass
                    else:
                        break
                else:
                    error_message = "Could not validate for an expected response model."
                    LOGGER.error(
                        "%s\nURL=%r\n" "response_models=%r\nresponse=%s",
                        error_message,
                        self.parse_config.configuration.downloadUrl,
                        response_model,
                        response,
                    )
                    raise OPTIMADEParseError(error_message)
            else:
                # No "endpoint" or unknown
                LOGGER.debug("No response_model, using Success response model.")
                try:
                    response_object = Success(**response.get("json", {}))
                except ValidationError as exc:
                    error_message = "Unknown or unparseable endpoint."
                    LOGGER.error(
                        "%s\nValidatonError: %s\n"
                        "URL=%r\nendpoint=%r\nresponse_model=%r\nresponse=%s",
                        error_message,
                        exc,
                        self.parse_config.configuration.downloadUrl,
                        self.parse_config.configuration.downloadUrl.endpoint,
                        response_model,
                        response,
                    )
                    raise OPTIMADEParseError(error_message) from exc

        result = OPTIMADEParseResult(
            model_config=self.parse_config.configuration.model_dump(),
            optimade_response_model=(
                response_object.__class__.__module__,
                response_object.__class__.__name__,
            ),
            optimade_response=response_object.model_dump(exclude_unset=True),
        )

        if (
            self.parse_config.configuration.optimade_config
            and self.parse_config.configuration.optimade_config.query_parameters
        ):
            result = result.model_copy(
                update={
                    "optimade_config": self.parse_config.configuration.optimade_config.model_copy(
                        update={
                            "query_parameters": self.parse_config.configuration.optimade_config.query_parameters.model_dump(
                                exclude_defaults=True,
                                exclude_unset=True,
                            )
                        }
                    )
                }
            )

        return result

get()

Request and parse an OPTIMADE response using OPT.

This method will be called through the strategy-specific endpoint of the OTE-API Services.

Configuration values provided in resource_config.configuration take precedence over the derived values from downloadUrl.

Workflow:

  1. Request OPTIMADE response.
  2. Parse as an OPTIMADE Python tools (OPT) pydantic response model.

Returns:

Type Description
OPTIMADEParseResult

An update model of key/value-pairs to be stored in the session-specific
OPTIMADEParseResult

context from services.
Source code in oteapi_optimade/strategies/parse.py 
 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
def get(self) -> OPTIMADEParseResult:
    """Request and parse an OPTIMADE response using OPT.

    This method will be called through the strategy-specific endpoint of the
    OTE-API Services.

    Configuration values provided in `resource_config.configuration` take
    precedence over the derived values from `downloadUrl`.

    Workflow:

    1. Request OPTIMADE response.
    2. Parse as an OPTIMADE Python tools (OPT) pydantic response model.

    Returns:
        An update model of key/value-pairs to be stored in the session-specific
        context from services.

    """
    if (
        self.parse_config.configuration.downloadUrl is None
        or self.parse_config.configuration.mediaType is None
    ):
        raise OPTIMADEParseError(
            "Missing downloadUrl or mediaType in configuration."
        )

    cache = DataCache(self.parse_config.configuration.datacache_config)
    if self.parse_config.configuration.downloadUrl in cache:
        response: dict[str, Any] = cache.get(
            self.parse_config.configuration.downloadUrl
        )
    elif (
        self.parse_config.configuration.datacache_config.accessKey
        and self.parse_config.configuration.datacache_config.accessKey in cache
    ):
        response = cache.get(
            self.parse_config.configuration.datacache_config.accessKey
        )
    else:
        download_config = self.parse_config.configuration.model_copy()
        download_output = create_strategy("download", download_config).get()
        response = {"json": json.loads(cache.get(download_output.pop("key")))}

    if (
        not response.get("ok", True)
        or (
            response.get("status_code", 200) < 200
            or response.get("status_code", 200) >= 300
        )
        or "errors" in response.get("json", {})
    ):
        # Error response
        try:
            response_object = ErrorResponse(**response.get("json", {}))
        except ValidationError as exc:
            error_message = "Could not validate an error response."
            LOGGER.error(
                "%s\nValidationError: " "%s\nresponse=%r",
                error_message,
                exc,
                response,
            )
            raise OPTIMADEParseError(error_message) from exc
    else:
        # Successful response
        response_model = (
            self.parse_config.configuration.downloadUrl.response_model()
        )
        LOGGER.debug("response_model=%r", response_model)
        if response_model:
            if not isinstance(response_model, tuple):
                response_model = (response_model,)

            for model_cls in response_model:
                try:
                    response_object = model_cls(**response.get("json", {}))
                except ValidationError:
                    pass
                else:
                    break
            else:
                error_message = "Could not validate for an expected response model."
                LOGGER.error(
                    "%s\nURL=%r\n" "response_models=%r\nresponse=%s",
                    error_message,
                    self.parse_config.configuration.downloadUrl,
                    response_model,
                    response,
                )
                raise OPTIMADEParseError(error_message)
        else:
            # No "endpoint" or unknown
            LOGGER.debug("No response_model, using Success response model.")
            try:
                response_object = Success(**response.get("json", {}))
            except ValidationError as exc:
                error_message = "Unknown or unparseable endpoint."
                LOGGER.error(
                    "%s\nValidatonError: %s\n"
                    "URL=%r\nendpoint=%r\nresponse_model=%r\nresponse=%s",
                    error_message,
                    exc,
                    self.parse_config.configuration.downloadUrl,
                    self.parse_config.configuration.downloadUrl.endpoint,
                    response_model,
                    response,
                )
                raise OPTIMADEParseError(error_message) from exc

    result = OPTIMADEParseResult(
        model_config=self.parse_config.configuration.model_dump(),
        optimade_response_model=(
            response_object.__class__.__module__,
            response_object.__class__.__name__,
        ),
        optimade_response=response_object.model_dump(exclude_unset=True),
    )

    if (
        self.parse_config.configuration.optimade_config
        and self.parse_config.configuration.optimade_config.query_parameters
    ):
        result = result.model_copy(
            update={
                "optimade_config": self.parse_config.configuration.optimade_config.model_copy(
                    update={
                        "query_parameters": self.parse_config.configuration.optimade_config.query_parameters.model_dump(
                            exclude_defaults=True,
                            exclude_unset=True,
                        )
                    }
                )
            }
        )

    return result

initialize()

Initialize strategy.

This method will be called through the /initialize endpoint of the OTE-API Services.

Returns:

Type Description
AttrDict

An update model of key/value-pairs to be stored in the session-specific
AttrDict

context from services.
Source code in oteapi_optimade/strategies/parse.py 
38
39
40
41
42
43
44
45
46
47
48
49
def initialize(self) -> AttrDict:
    """Initialize strategy.

    This method will be called through the `/initialize` endpoint of the OTE-API
    Services.

    Returns:
        An update model of key/value-pairs to be stored in the session-specific
        context from services.

    """
    return AttrDict()