OTE-API OPTIMADE Strategies

This page provides documentation for the oteapi_optimade.strategies submodule, where all the OTE-API OPTIMADE strategies are located.

These strategies will be available when setting up a server in an environment with oteapi-optimade installed.

filter

Demo filter strategy.

OPTIMADEFilterStrategy dataclass

Filter Strategy.

Implements strategies:

• ("filterType", "OPTIMADE")
• ("filterType", "optimade")
• ("filterType", "OPTiMaDe")

Source code in oteapi_optimade/strategies/filter.py

@dataclass
class OPTIMADEFilterStrategy:
    """Filter Strategy.

    **Implements strategies**:

    - `("filterType", "OPTIMADE")`
    - `("filterType", "optimade")`
    - `("filterType", "OPTiMaDe")`

    """

    filter_config: OPTIMADEFilterConfig

    def initialize(
        self, session: SessionUpdate | dict[str, Any] | None = None
    ) -> OPTIMADEFilterSession:
        """Initialize strategy.

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

        Configuration values, specifically URL query parameters, can be provided to the
        OPTIMADE resource strategy through this filter strategy.

        Workflow:

        1. Compile received information.
        2. Update session with compiled information.

        Parameters:
            session: A session-specific dictionary context.

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

        """
        if session and isinstance(session, dict):
            session = OPTIMADEFilterSession(**session)
        elif session and isinstance(session, SessionUpdate):
            session = OPTIMADEFilterSession(
                **session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        else:
            session = OPTIMADEFilterSession()

        if session.optimade_config:
            self.filter_config.configuration.update(
                session.optimade_config.model_dump(
                    exclude_defaults=True, exclude_unset=True
                )
            )

        optimade_config = self.filter_config.configuration.model_copy()

        if not optimade_config.query_parameters:
            optimade_config.query_parameters = OPTIMADEQueryParameters()

        if self.filter_config.query:
            LOGGER.debug("Setting filter from query.")
            optimade_config.query_parameters.filter = self.filter_config.query

        if self.filter_config.limit:
            LOGGER.debug("Setting page_limit from limit.")
            optimade_config.query_parameters.page_limit = self.filter_config.limit

        return session.model_copy(  # type: ignore[no-any-return]
            update={
                "optimade_config": optimade_config.model_copy(
                    update={
                        "query_parameters": optimade_config.query_parameters.model_dump(
                            exclude_defaults=True,
                            exclude_unset=True,
                        )
                    }
                )
            },
        )

    def get(
        self,
        session: dict[str, Any] | None = None,  # noqa: ARG002
    ) -> SessionUpdate:
        """Execute the strategy.

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

        Parameters:
            session: A session-specific dictionary context.

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

        """
        return SessionUpdate()

get(self, session=None)

Execute the strategy.

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

Parameters:

Name	Type	Description	Default
session	dict[str, Any] | None	A session-specific dictionary context.	None

Returns:

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

Source code in oteapi_optimade/strategies/filter.py

def get(
    self,
    session: dict[str, Any] | None = None,  # noqa: ARG002
) -> SessionUpdate:
    """Execute the strategy.

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

    Parameters:
        session: A session-specific dictionary context.

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

    """
    return SessionUpdate()

initialize(self, session=None)

Initialize strategy.

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

Configuration values, specifically URL query parameters, can be provided to the OPTIMADE resource strategy through this filter strategy.

Workflow:

1. Compile received information.
2. Update session with compiled information.

Parameters:

Name	Type	Description	Default
session	SessionUpdate | dict[str, Any] | None	A session-specific dictionary context.	None

Returns:

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

Source code in oteapi_optimade/strategies/filter.py

def initialize(
    self, session: SessionUpdate | dict[str, Any] | None = None
) -> OPTIMADEFilterSession:
    """Initialize strategy.

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

    Configuration values, specifically URL query parameters, can be provided to the
    OPTIMADE resource strategy through this filter strategy.

    Workflow:

    1. Compile received information.
    2. Update session with compiled information.

    Parameters:
        session: A session-specific dictionary context.

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

    """
    if session and isinstance(session, dict):
        session = OPTIMADEFilterSession(**session)
    elif session and isinstance(session, SessionUpdate):
        session = OPTIMADEFilterSession(
            **session.model_dump(exclude_defaults=True, exclude_unset=True)
        )
    else:
        session = OPTIMADEFilterSession()

    if session.optimade_config:
        self.filter_config.configuration.update(
            session.optimade_config.model_dump(
                exclude_defaults=True, exclude_unset=True
            )
        )

    optimade_config = self.filter_config.configuration.model_copy()

    if not optimade_config.query_parameters:
        optimade_config.query_parameters = OPTIMADEQueryParameters()

    if self.filter_config.query:
        LOGGER.debug("Setting filter from query.")
        optimade_config.query_parameters.filter = self.filter_config.query

    if self.filter_config.limit:
        LOGGER.debug("Setting page_limit from limit.")
        optimade_config.query_parameters.page_limit = self.filter_config.limit

    return session.model_copy(  # type: ignore[no-any-return]
        update={
            "optimade_config": optimade_config.model_copy(
                update={
                    "query_parameters": optimade_config.query_parameters.model_dump(
                        exclude_defaults=True,
                        exclude_unset=True,
                    )
                }
            )
        },
    )

parse

Demo strategy class for text/json.

OPTIMADEParseStrategy dataclass

Parse strategy for JSON.

Implements strategies:

• ("mediaType", "application/vnd.optimade+json")
• ("mediaType", "application/vnd.OPTIMADE+json")
• ("mediaType", "application/vnd.OPTiMaDe+json")
• ("mediaType", "application/vnd.optimade+JSON")
• ("mediaType", "application/vnd.OPTIMADE+JSON")
• ("mediaType", "application/vnd.OPTiMaDe+JSON")
• ("mediaType", "application/vnd.optimade")
• ("mediaType", "application/vnd.OPTIMADE")
• ("mediaType", "application/vnd.OPTiMaDe")

Source code in oteapi_optimade/strategies/parse.py

@dataclass
class OPTIMADEParseStrategy:
    """Parse strategy for JSON.

    **Implements strategies**:

    - `("mediaType", "application/vnd.optimade+json")`
    - `("mediaType", "application/vnd.OPTIMADE+json")`
    - `("mediaType", "application/vnd.OPTiMaDe+json")`
    - `("mediaType", "application/vnd.optimade+JSON")`
    - `("mediaType", "application/vnd.OPTIMADE+JSON")`
    - `("mediaType", "application/vnd.OPTiMaDe+JSON")`
    - `("mediaType", "application/vnd.optimade")`
    - `("mediaType", "application/vnd.OPTIMADE")`
    - `("mediaType", "application/vnd.OPTiMaDe")`

    """

    parse_config: OPTIMADEParseConfig

    def initialize(
        self,
        session: dict[str, Any] | None = None,  # noqa: ARG002
    ) -> SessionUpdate:
        """Initialize strategy.

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

        Parameters:
            session: A session-specific dictionary context.

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

        """
        return SessionUpdate()

    def get(
        self, session: SessionUpdate | dict[str, Any] | None = None
    ) -> OPTIMADEParseSession:
        """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.

        Parameters:
            session: A session-specific dictionary-like context.

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

        """
        if session and isinstance(session, dict):
            session = OPTIMADEParseSession(**session)
        elif session and isinstance(session, SessionUpdate):
            session = OPTIMADEParseSession(
                **session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        else:
            session = OPTIMADEParseSession()

        if session.optimade_config:
            self.parse_config.configuration.update(
                session.optimade_config.model_dump(
                    exclude_defaults=True, exclude_unset=True
                )
            )

        cache = DataCache(self.parse_config.configuration.datacache_config)
        if self.parse_config.downloadUrl in cache:
            response: dict[str, Any] = cache.get(self.parse_config.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.model_copy()
            session.update(
                create_strategy(StrategyType.DOWNLOAD, download_config).initialize(
                    session.model_dump(exclude_defaults=True, exclude_unset=True)
                )
            )
            session.update(
                create_strategy(StrategyType.DOWNLOAD, download_config).get(
                    session.model_dump(exclude_defaults=True, exclude_unset=True)
                )
            )

            response = {"json": json.loads(cache.get(session.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.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.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.downloadUrl,
                        self.parse_config.downloadUrl.endpoint,
                        response_model,
                        response,
                    )
                    raise OPTIMADEParseError(error_message) from exc

        session.optimade_response_model = (
            response_object.__class__.__module__,
            response_object.__class__.__name__,
        )
        session.optimade_response = response_object.model_dump(exclude_unset=True)

        if session.optimade_config and session.optimade_config.query_parameters:
            session = session.model_copy(
                update={
                    "optimade_config": session.optimade_config.model_copy(
                        update={
                            "query_parameters": session.optimade_config.query_parameters.model_dump(
                                exclude_defaults=True,
                                exclude_unset=True,
                            )
                        }
                    )
                }
            )

        if TYPE_CHECKING:  # pragma: no cover
            assert isinstance(session, OPTIMADEParseSession)  # nosec

        return session

get(self, session=None)

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.

Parameters:

Name	Type	Description	Default
session	SessionUpdate | dict[str, Any] | None	A session-specific dictionary-like context.	None

Returns:

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

Source code in oteapi_optimade/strategies/parse.py

def get(
    self, session: SessionUpdate | dict[str, Any] | None = None
) -> OPTIMADEParseSession:
    """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.

    Parameters:
        session: A session-specific dictionary-like context.

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

    """
    if session and isinstance(session, dict):
        session = OPTIMADEParseSession(**session)
    elif session and isinstance(session, SessionUpdate):
        session = OPTIMADEParseSession(
            **session.model_dump(exclude_defaults=True, exclude_unset=True)
        )
    else:
        session = OPTIMADEParseSession()

    if session.optimade_config:
        self.parse_config.configuration.update(
            session.optimade_config.model_dump(
                exclude_defaults=True, exclude_unset=True
            )
        )

    cache = DataCache(self.parse_config.configuration.datacache_config)
    if self.parse_config.downloadUrl in cache:
        response: dict[str, Any] = cache.get(self.parse_config.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.model_copy()
        session.update(
            create_strategy(StrategyType.DOWNLOAD, download_config).initialize(
                session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        )
        session.update(
            create_strategy(StrategyType.DOWNLOAD, download_config).get(
                session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        )

        response = {"json": json.loads(cache.get(session.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.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.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.downloadUrl,
                    self.parse_config.downloadUrl.endpoint,
                    response_model,
                    response,
                )
                raise OPTIMADEParseError(error_message) from exc

    session.optimade_response_model = (
        response_object.__class__.__module__,
        response_object.__class__.__name__,
    )
    session.optimade_response = response_object.model_dump(exclude_unset=True)

    if session.optimade_config and session.optimade_config.query_parameters:
        session = session.model_copy(
            update={
                "optimade_config": session.optimade_config.model_copy(
                    update={
                        "query_parameters": session.optimade_config.query_parameters.model_dump(
                            exclude_defaults=True,
                            exclude_unset=True,
                        )
                    }
                )
            }
        )

    if TYPE_CHECKING:  # pragma: no cover
        assert isinstance(session, OPTIMADEParseSession)  # nosec

    return session

initialize(self, session=None)

Initialize strategy.

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

Parameters:

Name	Type	Description	Default
session	dict[str, Any] | None	A session-specific dictionary context.	None

Returns:

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

Source code in oteapi_optimade/strategies/parse.py

def initialize(
    self,
    session: dict[str, Any] | None = None,  # noqa: ARG002
) -> SessionUpdate:
    """Initialize strategy.

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

    Parameters:
        session: A session-specific dictionary context.

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

    """
    return SessionUpdate()

resource

OPTIMADE resource strategy.

OPTIMADEResourceStrategy dataclass

OPTIMADE Resource Strategy.

Implements strategies:

• ("accessService", "optimade")
• ("accessService", "OPTIMADE")
• ("accessService", "OPTiMaDe")
• ("accessService", "optimade+dlite")
• ("accessService", "OPTIMADE+dlite")
• ("accessService", "OPTiMaDe+dlite")
• ("accessService", "optimade+DLite")
• ("accessService", "OPTIMADE+DLite")
• ("accessService", "OPTiMaDe+DLite")

Source code in oteapi_optimade/strategies/resource.py

@dataclass
class OPTIMADEResourceStrategy:
    """OPTIMADE Resource Strategy.

    **Implements strategies**:

    - `("accessService", "optimade")`
    - `("accessService", "OPTIMADE")`
    - `("accessService", "OPTiMaDe")`
    - `("accessService", "optimade+dlite")`
    - `("accessService", "OPTIMADE+dlite")`
    - `("accessService", "OPTiMaDe+dlite")`
    - `("accessService", "optimade+DLite")`
    - `("accessService", "OPTIMADE+DLite")`
    - `("accessService", "OPTiMaDe+DLite")`

    """

    resource_config: OPTIMADEResourceConfig

    def initialize(
        self, session: dict[str, Any] | None = None
    ) -> SessionUpdate | DLiteSessionUpdate:
        """Initialize strategy.

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

        Parameters:
            session: A session-specific dictionary context.

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

        """
        if use_dlite(
            self.resource_config.accessService,
            self.resource_config.configuration.use_dlite,
        ):
            return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
        return SessionUpdate()

    def get(
        self, session: SessionUpdate | dict[str, Any] | None = None
    ) -> OPTIMADEResourceSession:
        """Execute an OPTIMADE query to `accessUrl`.

        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 `accessUrl`.

        Workflow:
        1. Update configuration according to session.
        2. Deconstruct `accessUrl` (done partly by
           `oteapi_optimade.models.custom_types.OPTIMADEUrl`).
        3. Reconstruct the complete query URL.
        4. Send query.
        5. Store result in data cache.

        Parameters:
            session: A session-specific dictionary-like context.

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

        """
        if session and isinstance(session, dict):
            session = OPTIMADEResourceSession(**session)
        elif session and isinstance(session, SessionUpdate):
            session = OPTIMADEResourceSession(
                **session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        else:
            session = OPTIMADEResourceSession()

        if session.optimade_config:
            self.resource_config.configuration.update(
                session.optimade_config.model_dump(
                    exclude_defaults=True, exclude_unset=True
                )
            )

        optimade_endpoint = self.resource_config.accessUrl.endpoint or "structures"
        optimade_query = (
            self.resource_config.configuration.query_parameters
            or OPTIMADEQueryParameters()
        )
        LOGGER.debug("resource_config: %r", self.resource_config)

        if self.resource_config.accessUrl.query:
            parsed_query = parse_qs(self.resource_config.accessUrl.query)
            for field, value in parsed_query.items():
                # Only use the latest defined value for any parameter
                if field not in optimade_query.model_fields_set:
                    LOGGER.debug(
                        "Setting %r from accessUrl (value=%r)", field, value[-1]
                    )
                    setattr(optimade_query, field, value[-1])

        LOGGER.debug("optimade_query after update: %r", optimade_query)

        optimade_url = OPTIMADEUrl(
            f"{self.resource_config.accessUrl.base_url}"
            f"/{self.resource_config.accessUrl.version or 'v1'}"
            f"/{optimade_endpoint}?{optimade_query.generate_query_string()}"
        )
        LOGGER.debug("OPTIMADE URL to be requested: %s", optimade_url)

        # Set cache access key to the full OPTIMADE URL.
        self.resource_config.configuration.datacache_config.accessKey = optimade_url

        # Perform query
        response = requests.get(
            optimade_url,
            allow_redirects=True,
            timeout=(3, 27),  # timeout in seconds (connect, read)
        )

        if optimade_query.response_format and optimade_query.response_format != "json":
            error_message = (
                "Can only handle JSON responses for now. Requested response format: "
                f"{optimade_query.response_format!r}"
            )
            raise NotImplementedError(error_message)

        cache = DataCache(config=self.resource_config.configuration.datacache_config)
        cache.add(
            {
                "status_code": response.status_code,
                "ok": response.ok,
                "json": response.json(),
            }
        )

        parse_with_dlite = use_dlite(
            self.resource_config.accessService,
            self.resource_config.configuration.use_dlite,
        )

        parse_mediaType = (
            "application/vnd."
            f"{self.resource_config.accessService.split('+', maxsplit=1)[0]}"
        )
        if parse_with_dlite:
            parse_mediaType += "+DLite"
        elif optimade_query.response_format:
            parse_mediaType += f"+{optimade_query.response_format}"

        parse_config = {
            "downloadUrl": optimade_url,
            "mediaType": parse_mediaType,
            "configuration": {
                "datacache_config": self.resource_config.configuration.datacache_config,
            },
        }

        LOGGER.debug("parse_config: %r", parse_config)

        session.update(
            create_strategy(StrategyType.PARSE, parse_config).initialize(
                session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        )
        session.update(
            create_strategy(StrategyType.PARSE, parse_config).get(
                session.model_dump(exclude_defaults=True, exclude_unset=True)
            )
        )

        if not all(
            _ in session for _ in ("optimade_response", "optimade_response_model")
        ):
            base_error_message = (
                "Could not retrieve response from OPTIMADE parse strategy."
            )
            LOGGER.error(
                "%s\n"
                "optimade_response=%r\n"
                "optimade_response_model=%r\n"
                "session fields=%r",
                base_error_message,
                session.get("optimade_response"),
                session.get("optimade_response_model"),
                list(session.keys()),
            )
            raise OPTIMADEParseError(base_error_message)

        optimade_response_model_module, optimade_response_model_name = session.pop(
            "optimade_response_model"
        )
        optimade_response_dict = session.pop("optimade_response")

        # Parse response using the provided model
        try:
            optimade_response_model: type[OPTIMADEResponse] = getattr(
                importlib.import_module(optimade_response_model_module),
                optimade_response_model_name,
            )
            optimade_response = optimade_response_model(**optimade_response_dict)
        except (ImportError, AttributeError) as exc:
            base_error_message = "Could not import the response model."
            LOGGER.error(
                "%s\n"
                "ImportError: %s\n"
                "optimade_response_model_module=%r\n"
                "optimade_response_model_name=%r",
                base_error_message,
                exc,
                optimade_response_model_module,
                optimade_response_model_name,
            )
            raise OPTIMADEParseError(base_error_message) from exc
        except ValidationError as exc:
            base_error_message = "Could not validate the response model."
            LOGGER.error(
                "%s\n"
                "ValidationError: %s\n"
                "optimade_response_model_module=%r\n"
                "optimade_response_model_name=%r",
                base_error_message,
                exc,
                optimade_response_model_module,
                optimade_response_model_name,
            )
            raise OPTIMADEParseError(base_error_message) from exc

        if isinstance(optimade_response, ErrorResponse):
            optimade_resources = optimade_response.errors
            session.optimade_resource_model = (
                f"{OptimadeError.__module__}:OptimadeError"
            )
        elif isinstance(optimade_response, ReferenceResponseMany):
            optimade_resources = [
                (
                    Reference(entry).as_dict
                    if isinstance(entry, dict)
                    else Reference(entry.model_dump()).as_dict
                )
                for entry in optimade_response.data
            ]
            session.optimade_resource_model = f"{Reference.__module__}:Reference"
        elif isinstance(optimade_response, ReferenceResponseOne):
            optimade_resources = [
                (
                    Reference(optimade_response.data).as_dict
                    if isinstance(optimade_response.data, dict)
                    else Reference(optimade_response.data.model_dump()).as_dict
                )
            ]
            session.optimade_resource_model = f"{Reference.__module__}:Reference"
        elif isinstance(optimade_response, StructureResponseMany):
            optimade_resources = [
                (
                    Structure(entry).as_dict
                    if isinstance(entry, dict)
                    else Structure(entry.model_dump()).as_dict
                )
                for entry in optimade_response.data
            ]
            session.optimade_resource_model = f"{Structure.__module__}:Structure"
        elif isinstance(optimade_response, StructureResponseOne):
            optimade_resources = [
                (
                    Structure(optimade_response.data).as_dict
                    if isinstance(optimade_response.data, dict)
                    else Structure(optimade_response.data.model_dump()).as_dict
                )
            ]
            session.optimade_resource_model = f"{Structure.__module__}:Structure"
        else:
            LOGGER.error(
                "Could not parse response as errors, references or structures. "
                "Response:\n%r",
                optimade_response,
            )
            error_message = (
                "Could not retrieve errors, references or structures from response "
                f"from {optimade_url}. It could be a valid OPTIMADE API response, "
                "however it may not be supported by OTEAPI-OPTIMADE. It may also be an "
                "invalid response completely."
            )
            raise OPTIMADEParseError(error_message)

        session.optimade_resources = [
            resource if isinstance(resource, dict) else resource.model_dump()
            for resource in optimade_resources
        ]

        if session.optimade_config and session.optimade_config.query_parameters:
            session = session.model_copy(
                update={
                    "optimade_config": session.optimade_config.model_copy(
                        update={
                            "query_parameters": session.optimade_config.query_parameters.model_dump(
                                exclude_defaults=True,
                                exclude_unset=True,
                            )
                        }
                    )
                }
            )

        if TYPE_CHECKING:  # pragma: no cover
            assert isinstance(session, OPTIMADEResourceSession)  # nosec

        return session

get(self, session=None)

Execute an OPTIMADE query to accessUrl.

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 accessUrl.

Workflow: 1. Update configuration according to session. 2. Deconstruct accessUrl (done partly by oteapi_optimade.models.custom_types.OPTIMADEUrl). 3. Reconstruct the complete query URL. 4. Send query. 5. Store result in data cache.

Parameters:

Name	Type	Description	Default
session	SessionUpdate | dict[str, Any] | None	A session-specific dictionary-like context.	None

Returns:

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

Source code in oteapi_optimade/strategies/resource.py

def get(
    self, session: SessionUpdate | dict[str, Any] | None = None
) -> OPTIMADEResourceSession:
    """Execute an OPTIMADE query to `accessUrl`.

    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 `accessUrl`.

    Workflow:
    1. Update configuration according to session.
    2. Deconstruct `accessUrl` (done partly by
       `oteapi_optimade.models.custom_types.OPTIMADEUrl`).
    3. Reconstruct the complete query URL.
    4. Send query.
    5. Store result in data cache.

    Parameters:
        session: A session-specific dictionary-like context.

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

    """
    if session and isinstance(session, dict):
        session = OPTIMADEResourceSession(**session)
    elif session and isinstance(session, SessionUpdate):
        session = OPTIMADEResourceSession(
            **session.model_dump(exclude_defaults=True, exclude_unset=True)
        )
    else:
        session = OPTIMADEResourceSession()

    if session.optimade_config:
        self.resource_config.configuration.update(
            session.optimade_config.model_dump(
                exclude_defaults=True, exclude_unset=True
            )
        )

    optimade_endpoint = self.resource_config.accessUrl.endpoint or "structures"
    optimade_query = (
        self.resource_config.configuration.query_parameters
        or OPTIMADEQueryParameters()
    )
    LOGGER.debug("resource_config: %r", self.resource_config)

    if self.resource_config.accessUrl.query:
        parsed_query = parse_qs(self.resource_config.accessUrl.query)
        for field, value in parsed_query.items():
            # Only use the latest defined value for any parameter
            if field not in optimade_query.model_fields_set:
                LOGGER.debug(
                    "Setting %r from accessUrl (value=%r)", field, value[-1]
                )
                setattr(optimade_query, field, value[-1])

    LOGGER.debug("optimade_query after update: %r", optimade_query)

    optimade_url = OPTIMADEUrl(
        f"{self.resource_config.accessUrl.base_url}"
        f"/{self.resource_config.accessUrl.version or 'v1'}"
        f"/{optimade_endpoint}?{optimade_query.generate_query_string()}"
    )
    LOGGER.debug("OPTIMADE URL to be requested: %s", optimade_url)

    # Set cache access key to the full OPTIMADE URL.
    self.resource_config.configuration.datacache_config.accessKey = optimade_url

    # Perform query
    response = requests.get(
        optimade_url,
        allow_redirects=True,
        timeout=(3, 27),  # timeout in seconds (connect, read)
    )

    if optimade_query.response_format and optimade_query.response_format != "json":
        error_message = (
            "Can only handle JSON responses for now. Requested response format: "
            f"{optimade_query.response_format!r}"
        )
        raise NotImplementedError(error_message)

    cache = DataCache(config=self.resource_config.configuration.datacache_config)
    cache.add(
        {
            "status_code": response.status_code,
            "ok": response.ok,
            "json": response.json(),
        }
    )

    parse_with_dlite = use_dlite(
        self.resource_config.accessService,
        self.resource_config.configuration.use_dlite,
    )

    parse_mediaType = (
        "application/vnd."
        f"{self.resource_config.accessService.split('+', maxsplit=1)[0]}"
    )
    if parse_with_dlite:
        parse_mediaType += "+DLite"
    elif optimade_query.response_format:
        parse_mediaType += f"+{optimade_query.response_format}"

    parse_config = {
        "downloadUrl": optimade_url,
        "mediaType": parse_mediaType,
        "configuration": {
            "datacache_config": self.resource_config.configuration.datacache_config,
        },
    }

    LOGGER.debug("parse_config: %r", parse_config)

    session.update(
        create_strategy(StrategyType.PARSE, parse_config).initialize(
            session.model_dump(exclude_defaults=True, exclude_unset=True)
        )
    )
    session.update(
        create_strategy(StrategyType.PARSE, parse_config).get(
            session.model_dump(exclude_defaults=True, exclude_unset=True)
        )
    )

    if not all(
        _ in session for _ in ("optimade_response", "optimade_response_model")
    ):
        base_error_message = (
            "Could not retrieve response from OPTIMADE parse strategy."
        )
        LOGGER.error(
            "%s\n"
            "optimade_response=%r\n"
            "optimade_response_model=%r\n"
            "session fields=%r",
            base_error_message,
            session.get("optimade_response"),
            session.get("optimade_response_model"),
            list(session.keys()),
        )
        raise OPTIMADEParseError(base_error_message)

    optimade_response_model_module, optimade_response_model_name = session.pop(
        "optimade_response_model"
    )
    optimade_response_dict = session.pop("optimade_response")

    # Parse response using the provided model
    try:
        optimade_response_model: type[OPTIMADEResponse] = getattr(
            importlib.import_module(optimade_response_model_module),
            optimade_response_model_name,
        )
        optimade_response = optimade_response_model(**optimade_response_dict)
    except (ImportError, AttributeError) as exc:
        base_error_message = "Could not import the response model."
        LOGGER.error(
            "%s\n"
            "ImportError: %s\n"
            "optimade_response_model_module=%r\n"
            "optimade_response_model_name=%r",
            base_error_message,
            exc,
            optimade_response_model_module,
            optimade_response_model_name,
        )
        raise OPTIMADEParseError(base_error_message) from exc
    except ValidationError as exc:
        base_error_message = "Could not validate the response model."
        LOGGER.error(
            "%s\n"
            "ValidationError: %s\n"
            "optimade_response_model_module=%r\n"
            "optimade_response_model_name=%r",
            base_error_message,
            exc,
            optimade_response_model_module,
            optimade_response_model_name,
        )
        raise OPTIMADEParseError(base_error_message) from exc

    if isinstance(optimade_response, ErrorResponse):
        optimade_resources = optimade_response.errors
        session.optimade_resource_model = (
            f"{OptimadeError.__module__}:OptimadeError"
        )
    elif isinstance(optimade_response, ReferenceResponseMany):
        optimade_resources = [
            (
                Reference(entry).as_dict
                if isinstance(entry, dict)
                else Reference(entry.model_dump()).as_dict
            )
            for entry in optimade_response.data
        ]
        session.optimade_resource_model = f"{Reference.__module__}:Reference"
    elif isinstance(optimade_response, ReferenceResponseOne):
        optimade_resources = [
            (
                Reference(optimade_response.data).as_dict
                if isinstance(optimade_response.data, dict)
                else Reference(optimade_response.data.model_dump()).as_dict
            )
        ]
        session.optimade_resource_model = f"{Reference.__module__}:Reference"
    elif isinstance(optimade_response, StructureResponseMany):
        optimade_resources = [
            (
                Structure(entry).as_dict
                if isinstance(entry, dict)
                else Structure(entry.model_dump()).as_dict
            )
            for entry in optimade_response.data
        ]
        session.optimade_resource_model = f"{Structure.__module__}:Structure"
    elif isinstance(optimade_response, StructureResponseOne):
        optimade_resources = [
            (
                Structure(optimade_response.data).as_dict
                if isinstance(optimade_response.data, dict)
                else Structure(optimade_response.data.model_dump()).as_dict
            )
        ]
        session.optimade_resource_model = f"{Structure.__module__}:Structure"
    else:
        LOGGER.error(
            "Could not parse response as errors, references or structures. "
            "Response:\n%r",
            optimade_response,
        )
        error_message = (
            "Could not retrieve errors, references or structures from response "
            f"from {optimade_url}. It could be a valid OPTIMADE API response, "
            "however it may not be supported by OTEAPI-OPTIMADE. It may also be an "
            "invalid response completely."
        )
        raise OPTIMADEParseError(error_message)

    session.optimade_resources = [
        resource if isinstance(resource, dict) else resource.model_dump()
        for resource in optimade_resources
    ]

    if session.optimade_config and session.optimade_config.query_parameters:
        session = session.model_copy(
            update={
                "optimade_config": session.optimade_config.model_copy(
                    update={
                        "query_parameters": session.optimade_config.query_parameters.model_dump(
                            exclude_defaults=True,
                            exclude_unset=True,
                        )
                    }
                )
            }
        )

    if TYPE_CHECKING:  # pragma: no cover
        assert isinstance(session, OPTIMADEResourceSession)  # nosec

    return session

initialize(self, session=None)

Initialize strategy.

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

Parameters:

Name	Type	Description	Default
session	dict[str, Any] | None	A session-specific dictionary context.	None

Returns:

Type	Description
SessionUpdate | DLiteSessionUpdate	An update model of key/value-pairs to be stored in the session-specific context from services.

Source code in oteapi_optimade/strategies/resource.py

def initialize(
    self, session: dict[str, Any] | None = None
) -> SessionUpdate | DLiteSessionUpdate:
    """Initialize strategy.

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

    Parameters:
        session: A session-specific dictionary context.

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

    """
    if use_dlite(
        self.resource_config.accessService,
        self.resource_config.configuration.use_dlite,
    ):
        return DLiteSessionUpdate(collection_id=get_collection(session).uuid)
    return SessionUpdate()

use_dlite(access_service, use_dlite_flag)

Determine whether DLite should be utilized in the Resource strategy.

Parameters:

Name	Type	Description	Default
access_service	str	The accessService value from the resource's configuration.	required
use_dlite_flag	bool	The strategy-specific use_dlite configuration option.	required

Returns:

Type	Description
bool	Based on the accessService value, then whether DLite should be used or not.

Source code in oteapi_optimade/strategies/resource.py

def use_dlite(access_service: str, use_dlite_flag: bool) -> bool:
    """Determine whether DLite should be utilized in the Resource strategy.

    Parameters:
        access_service: The accessService value from the resource's configuration.
        use_dlite_flag: The strategy-specific `use_dlite` configuration option.

    Returns:
        Based on the accessService value, then whether DLite should be used or not.

    """
    if (
        any(dlite_form in access_service for dlite_form in ["DLite", "dlite"])
        or use_dlite_flag
    ):
        if oteapi_dlite_version is None:
            error_message = (
                "OTEAPI-DLite is not found on the system. This is required to use "
                "DLite with the OTEAPI-OPTIMADE strategies."
            )
            raise MissingDependency(error_message)
        return True
    return False