Skip to content

entry_collections

CONFIG: ServerConfig = ServerConfig() module-attribute

This singleton loads the config from a hierarchy of sources (see customise_sources) and makes it importable in the server code.

NoneType = type(None) module-attribute

Attributes

Bases: BaseModel

Members of the attributes object ("attributes") represent information about the resource object in which it's defined. The keys for Attributes MUST NOT be: relationships links id type

Source code in optimade/models/jsonapi.py 
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
class Attributes(BaseModel):
    """
    Members of the attributes object ("attributes\") represent information about the resource object in which it's defined.
    The keys for Attributes MUST NOT be:
        relationships
        links
        id
        type
    """

    model_config = ConfigDict(extra="allow")

    @model_validator(mode="after")
    def check_illegal_attributes_fields(self) -> "Attributes":
        illegal_fields = ("relationships", "links", "id", "type")
        for field in illegal_fields:
            if hasattr(self, field):
                raise ValueError(
                    f"{illegal_fields} MUST NOT be fields under Attributes"
                )
        return self

BadRequest

Bases: OptimadeHTTPException

400 Bad Request

Source code in optimade/exceptions.py 
57
58
59
60
61
class BadRequest(OptimadeHTTPException):
    """400 Bad Request"""

    status_code: int = 400
    title: str = "Bad Request"

BaseResourceMapper

Generic Resource Mapper that defines and performs the mapping between objects in the database and the resource objects defined by the specification.

Attributes:

Name Type Description
ALIASES tuple[tuple[str, str], ...]

a tuple of aliases between OPTIMADE field names and the field names in the database , e.g. (("elements", "custom_elements_field")).
LENGTH_ALIASES tuple[tuple[str, str], ...]

a tuple of aliases between a field name and another field that defines its length, to be used when querying, e.g. (("elements", "nelements")). e.g. (("elements", "custom_elements_field")).
ENTRY_RESOURCE_CLASS type[EntryResource]

The entry type that this mapper corresponds to.
PROVIDER_FIELDS tuple[str, ...]

a tuple of extra field names that this mapper should support when querying with the database prefix.
TOP_LEVEL_NON_ATTRIBUTES_FIELDS set[str]

the set of top-level field names common to all endpoints.
SUPPORTED_PREFIXES set[str]

The set of prefixes registered by this mapper.
ALL_ATTRIBUTES set[str]

The set of attributes defined across the entry resource class and the server configuration.
ENTRY_RESOURCE_ATTRIBUTES dict[str, Any]

A dictionary of attributes and their definitions defined by the schema of the entry resource class.
ENDPOINT str

The expected endpoint name for this resource, as defined by the type in the schema of the entry resource class.
Source code in optimade/server/mappers/entries.py 
 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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
class BaseResourceMapper:
    """Generic Resource Mapper that defines and performs the mapping
    between objects in the database and the resource objects defined by
    the specification.

    Attributes:
        ALIASES: a tuple of aliases between
            OPTIMADE field names and the field names in the database ,
            e.g. `(("elements", "custom_elements_field"))`.
        LENGTH_ALIASES: a tuple of aliases between
            a field name and another field that defines its length, to be used
            when querying, e.g. `(("elements", "nelements"))`.
            e.g. `(("elements", "custom_elements_field"))`.
        ENTRY_RESOURCE_CLASS: The entry type that this mapper corresponds to.
        PROVIDER_FIELDS: a tuple of extra field names that this
            mapper should support when querying with the database prefix.
        TOP_LEVEL_NON_ATTRIBUTES_FIELDS: the set of top-level
            field names common to all endpoints.
        SUPPORTED_PREFIXES: The set of prefixes registered by this mapper.
        ALL_ATTRIBUTES: The set of attributes defined across the entry
            resource class and the server configuration.
        ENTRY_RESOURCE_ATTRIBUTES: A dictionary of attributes and their definitions
            defined by the schema of the entry resource class.
        ENDPOINT: The expected endpoint name for this resource, as defined by
            the `type` in the schema of the entry resource class.

    """

    try:
        from optimade.server.data import providers as PROVIDERS  # type: ignore
    except (ImportError, ModuleNotFoundError):
        PROVIDERS = {}

    KNOWN_PROVIDER_PREFIXES: set[str] = {
        prov["id"] for prov in PROVIDERS.get("data", [])
    }
    ALIASES: tuple[tuple[str, str], ...] = ()
    LENGTH_ALIASES: tuple[tuple[str, str], ...] = ()
    PROVIDER_FIELDS: tuple[str, ...] = ()
    ENTRY_RESOURCE_CLASS: type[EntryResource] = EntryResource
    RELATIONSHIP_ENTRY_TYPES: set[str] = {"references", "structures"}
    TOP_LEVEL_NON_ATTRIBUTES_FIELDS: set[str] = {"id", "type", "relationships", "links"}

    @classmethod
    @lru_cache(maxsize=NUM_ENTRY_TYPES)
    def all_aliases(cls) -> Iterable[tuple[str, str]]:
        """Returns all of the associated aliases for this entry type,
        including those defined by the server config. The first member
        of each tuple is the OPTIMADE-compliant field name, the second
        is the backend-specific field name.

        Returns:
            A tuple of alias tuples.

        """
        from optimade.server.config import CONFIG

        return (
            tuple(
                (f"_{CONFIG.provider.prefix}_{field}", field)
                if not field.startswith("_")
                else (field, field)
                for field in CONFIG.provider_fields.get(cls.ENDPOINT, [])
                if isinstance(field, str)
            )
            + tuple(
                (f"_{CONFIG.provider.prefix}_{field['name']}", field["name"])
                if not field["name"].startswith("_")
                else (field["name"], field["name"])
                for field in CONFIG.provider_fields.get(cls.ENDPOINT, [])
                if isinstance(field, dict)
            )
            + tuple(
                (f"_{CONFIG.provider.prefix}_{field}", field)
                if not field.startswith("_")
                else (field, field)
                for field in cls.PROVIDER_FIELDS
            )
            + tuple(CONFIG.aliases.get(cls.ENDPOINT, {}).items())
            + cls.ALIASES
        )

    @classproperty
    @lru_cache(maxsize=1)
    def SUPPORTED_PREFIXES(cls) -> set[str]:
        """A set of prefixes handled by this entry type.

        !!! note
            This implementation only includes the provider prefix,
            but in the future this property may be extended to include other
            namespaces (for serving fields from, e.g., other providers or
            domain-specific terms).

        """
        from optimade.server.config import CONFIG

        return {CONFIG.provider.prefix}

    @classproperty
    def ALL_ATTRIBUTES(cls) -> set[str]:
        """Returns all attributes served by this entry."""
        from optimade.server.config import CONFIG

        return (
            set(cls.ENTRY_RESOURCE_ATTRIBUTES)
            .union(
                cls.get_optimade_field(field)
                for field in CONFIG.provider_fields.get(cls.ENDPOINT, ())
                if isinstance(field, str)
            )
            .union(
                cls.get_optimade_field(field["name"])
                for field in CONFIG.provider_fields.get(cls.ENDPOINT, ())
                if isinstance(field, dict)
            )
            .union({cls.get_optimade_field(field) for field in cls.PROVIDER_FIELDS})
        )

    @classproperty
    def ENTRY_RESOURCE_ATTRIBUTES(cls) -> dict[str, Any]:
        """Returns the dictionary of attributes defined by the underlying entry resource class."""
        from optimade.server.schemas import retrieve_queryable_properties

        return retrieve_queryable_properties(cls.ENTRY_RESOURCE_CLASS)

    @classproperty
    @lru_cache(maxsize=NUM_ENTRY_TYPES)
    def ENDPOINT(cls) -> str:
        """Returns the expected endpoint for this mapper, corresponding
        to the `type` property of the resource class.

        """
        endpoint = cls.ENTRY_RESOURCE_CLASS.model_fields["type"].default
        if not endpoint and not isinstance(endpoint, str):
            raise ValueError("Type not set for this entry type!")
        return endpoint

    @classmethod
    @lru_cache(maxsize=NUM_ENTRY_TYPES)
    def all_length_aliases(cls) -> tuple[tuple[str, str], ...]:
        """Returns all of the associated length aliases for this class,
        including those defined by the server config.

        Returns:
            A tuple of length alias tuples.

        """
        from optimade.server.config import CONFIG

        return cls.LENGTH_ALIASES + tuple(
            CONFIG.length_aliases.get(cls.ENDPOINT, {}).items()
        )

    @classmethod
    @lru_cache(maxsize=128)
    def length_alias_for(cls, field: str) -> Optional[str]:
        """Returns the length alias for the particular field,
        or `None` if no such alias is found.

        Parameters:
            field: OPTIMADE field name.

        Returns:
            Aliased field as found in [`all_length_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_length_aliases].

        """
        return dict(cls.all_length_aliases()).get(field, None)

    @classmethod
    @lru_cache(maxsize=128)
    def get_backend_field(cls, optimade_field: str) -> str:
        """Return the field name configured for the particular
        underlying database for the passed OPTIMADE field name, that would
        be used in an API filter.

        Aliases are read from
        [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

        If a dot-separated OPTIMADE field is provided, e.g., `species.mass`, only the first part will be mapped.
        This means for an (OPTIMADE, DB) alias of (`species`, `kinds`), `get_backend_fields("species.mass")`
        will return `kinds.mass`.

        Arguments:
            optimade_field: The OPTIMADE field to attempt to map to the backend-specific field.

        Examples:
            >>> get_backend_field("chemical_formula_anonymous")
            'formula_anon'
            >>> get_backend_field("formula_anon")
            'formula_anon'
            >>> get_backend_field("_exmpl_custom_provider_field")
            'custom_provider_field'

        Returns:
            The mapped field name to be used in the query to the backend.

        """
        split = optimade_field.split(".")
        alias = dict(cls.all_aliases()).get(split[0], None)
        if alias is not None:
            return alias + ("." + ".".join(split[1:]) if len(split) > 1 else "")
        return optimade_field

    @classmethod
    @lru_cache(maxsize=128)
    def alias_for(cls, field: str) -> str:
        """Return aliased field name.

        !!! warning "Deprecated"
            This method is deprecated could be removed without further warning. Please use
            [`get_backend_field()`][optimade.server.mappers.entries.BaseResourceMapper.get_backend_field].

        Parameters:
            field: OPTIMADE field name.

        Returns:
            Aliased field as found in [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

        """
        warnings.warn(
            "The `.alias_for(...)` method is deprecated, please use `.get_backend_field(...)`.",
            DeprecationWarning,
        )
        return cls.get_backend_field(field)

    @classmethod
    @lru_cache(maxsize=128)
    def get_optimade_field(cls, backend_field: str) -> str:
        """Return the corresponding OPTIMADE field name for the underlying database field,
        ready to be used to construct the OPTIMADE-compliant JSON response.

        Aliases are read from
        [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

        Arguments:
            backend_field: The backend field to attempt to map to an OPTIMADE field.

        Examples:
            >>> get_optimade_field("chemical_formula_anonymous")
            'chemical_formula_anonymous'
            >>> get_optimade_field("formula_anon")
            'chemical_formula_anonymous'
            >>> get_optimade_field("custom_provider_field")
            '_exmpl_custom_provider_field'

        Returns:
            The mapped field name to be used in an OPTIMADE-compliant response.

        """
        return {alias: real for real, alias in cls.all_aliases()}.get(
            backend_field, backend_field
        )

    @classmethod
    @lru_cache(maxsize=128)
    def alias_of(cls, field: str) -> str:
        """Return de-aliased field name, if it exists,
        otherwise return the input field name.

        !!! warning "Deprecated"
            This method is deprecated could be removed without further warning. Please use
            [`get_optimade_field()`][optimade.server.mappers.entries.BaseResourceMapper.get_optimade_field].

        Parameters:
            field: Field name to be de-aliased.

        Returns:
            De-aliased field name, falling back to returning `field`.

        """
        warnings.warn(
            "The `.alias_of(...)` method is deprecated, please use `.get_optimade_field(...)`.",
            DeprecationWarning,
        )
        return cls.get_optimade_field(field)

    @classmethod
    @lru_cache(maxsize=NUM_ENTRY_TYPES)
    def get_required_fields(cls) -> set:
        """Get REQUIRED response fields.

        Returns:
            REQUIRED response fields.

        """
        return cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS

    @classmethod
    def map_back(cls, doc: dict) -> dict:
        """Map properties from MongoDB to OPTIMADE.

        Starting from a MongoDB document `doc`, map the DB fields to the corresponding OPTIMADE fields.
        Then, the fields are all added to the top-level field "attributes",
        with the exception of other top-level fields, defined in `cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS`.
        All fields not in `cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS` + "attributes" will be removed.
        Finally, the `type` is given the value of the specified `cls.ENDPOINT`.

        Parameters:
            doc: A resource object in MongoDB format.

        Returns:
            A resource object in OPTIMADE format.

        """
        mapping = ((real, alias) for alias, real in cls.all_aliases())
        newdoc = {}
        reals = {real for _, real in cls.all_aliases()}
        for key in doc:
            if key not in reals:
                newdoc[key] = doc[key]
        for real, alias in mapping:
            if real in doc:
                newdoc[alias] = doc[real]

        if "attributes" in newdoc:
            raise Exception("Will overwrite doc field!")
        attributes = newdoc.copy()

        for field in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS:
            value = attributes.pop(field, None)
            if value is not None:
                newdoc[field] = value
        for field in list(newdoc.keys()):
            if field not in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS:
                del newdoc[field]

        newdoc["type"] = cls.ENDPOINT
        newdoc["attributes"] = attributes

        return newdoc

    @classmethod
    def deserialize(
        cls, results: Union[dict, Iterable[dict]]
    ) -> Union[list[EntryResource], EntryResource]:
        """Converts the raw database entries for this class into serialized models,
        mapping the data along the way.

        """
        if isinstance(results, dict):
            return cls.ENTRY_RESOURCE_CLASS(**cls.map_back(results))

        return [cls.ENTRY_RESOURCE_CLASS(**cls.map_back(doc)) for doc in results]

ALL_ATTRIBUTES()

Returns all attributes served by this entry.

Source code in optimade/server/mappers/entries.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
@classproperty
def ALL_ATTRIBUTES(cls) -> set[str]:
    """Returns all attributes served by this entry."""
    from optimade.server.config import CONFIG

    return (
        set(cls.ENTRY_RESOURCE_ATTRIBUTES)
        .union(
            cls.get_optimade_field(field)
            for field in CONFIG.provider_fields.get(cls.ENDPOINT, ())
            if isinstance(field, str)
        )
        .union(
            cls.get_optimade_field(field["name"])
            for field in CONFIG.provider_fields.get(cls.ENDPOINT, ())
            if isinstance(field, dict)
        )
        .union({cls.get_optimade_field(field) for field in cls.PROVIDER_FIELDS})
    )

ENDPOINT() cached

Returns the expected endpoint for this mapper, corresponding to the type property of the resource class.

Source code in optimade/server/mappers/entries.py 
161
162
163
164
165
166
167
168
169
170
171
@classproperty
@lru_cache(maxsize=NUM_ENTRY_TYPES)
def ENDPOINT(cls) -> str:
    """Returns the expected endpoint for this mapper, corresponding
    to the `type` property of the resource class.

    """
    endpoint = cls.ENTRY_RESOURCE_CLASS.model_fields["type"].default
    if not endpoint and not isinstance(endpoint, str):
        raise ValueError("Type not set for this entry type!")
    return endpoint

ENTRY_RESOURCE_ATTRIBUTES()

Returns the dictionary of attributes defined by the underlying entry resource class.

Source code in optimade/server/mappers/entries.py 
154
155
156
157
158
159
@classproperty
def ENTRY_RESOURCE_ATTRIBUTES(cls) -> dict[str, Any]:
    """Returns the dictionary of attributes defined by the underlying entry resource class."""
    from optimade.server.schemas import retrieve_queryable_properties

    return retrieve_queryable_properties(cls.ENTRY_RESOURCE_CLASS)

SUPPORTED_PREFIXES() cached

A set of prefixes handled by this entry type.

Note

This implementation only includes the provider prefix, but in the future this property may be extended to include other namespaces (for serving fields from, e.g., other providers or domain-specific terms).

Source code in optimade/server/mappers/entries.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
@classproperty
@lru_cache(maxsize=1)
def SUPPORTED_PREFIXES(cls) -> set[str]:
    """A set of prefixes handled by this entry type.

    !!! note
        This implementation only includes the provider prefix,
        but in the future this property may be extended to include other
        namespaces (for serving fields from, e.g., other providers or
        domain-specific terms).

    """
    from optimade.server.config import CONFIG

    return {CONFIG.provider.prefix}

alias_for(field) cached classmethod

Return aliased field name.

Deprecated

This method is deprecated could be removed without further warning. Please use get_backend_field().

Parameters:

Name Type Description Default
field str

OPTIMADE field name.

 required

Returns:

Type Description
str

Aliased field as found in all_aliases().
Source code in optimade/server/mappers/entries.py 
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
@classmethod
@lru_cache(maxsize=128)
def alias_for(cls, field: str) -> str:
    """Return aliased field name.

    !!! warning "Deprecated"
        This method is deprecated could be removed without further warning. Please use
        [`get_backend_field()`][optimade.server.mappers.entries.BaseResourceMapper.get_backend_field].

    Parameters:
        field: OPTIMADE field name.

    Returns:
        Aliased field as found in [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

    """
    warnings.warn(
        "The `.alias_for(...)` method is deprecated, please use `.get_backend_field(...)`.",
        DeprecationWarning,
    )
    return cls.get_backend_field(field)

alias_of(field) cached classmethod

Return de-aliased field name, if it exists, otherwise return the input field name.

Deprecated

This method is deprecated could be removed without further warning. Please use get_optimade_field().

Parameters:

Name Type Description Default
field str

Field name to be de-aliased.

 required

Returns:

Type Description
str

De-aliased field name, falling back to returning field.
Source code in optimade/server/mappers/entries.py 
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
@classmethod
@lru_cache(maxsize=128)
def alias_of(cls, field: str) -> str:
    """Return de-aliased field name, if it exists,
    otherwise return the input field name.

    !!! warning "Deprecated"
        This method is deprecated could be removed without further warning. Please use
        [`get_optimade_field()`][optimade.server.mappers.entries.BaseResourceMapper.get_optimade_field].

    Parameters:
        field: Field name to be de-aliased.

    Returns:
        De-aliased field name, falling back to returning `field`.

    """
    warnings.warn(
        "The `.alias_of(...)` method is deprecated, please use `.get_optimade_field(...)`.",
        DeprecationWarning,
    )
    return cls.get_optimade_field(field)

all_aliases() cached classmethod

Returns all of the associated aliases for this entry type, including those defined by the server config. The first member of each tuple is the OPTIMADE-compliant field name, the second is the backend-specific field name.

Returns:

Type Description
Iterable[tuple[str, str]]

A tuple of alias tuples.
Source code in optimade/server/mappers/entries.py 
 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
@classmethod
@lru_cache(maxsize=NUM_ENTRY_TYPES)
def all_aliases(cls) -> Iterable[tuple[str, str]]:
    """Returns all of the associated aliases for this entry type,
    including those defined by the server config. The first member
    of each tuple is the OPTIMADE-compliant field name, the second
    is the backend-specific field name.

    Returns:
        A tuple of alias tuples.

    """
    from optimade.server.config import CONFIG

    return (
        tuple(
            (f"_{CONFIG.provider.prefix}_{field}", field)
            if not field.startswith("_")
            else (field, field)
            for field in CONFIG.provider_fields.get(cls.ENDPOINT, [])
            if isinstance(field, str)
        )
        + tuple(
            (f"_{CONFIG.provider.prefix}_{field['name']}", field["name"])
            if not field["name"].startswith("_")
            else (field["name"], field["name"])
            for field in CONFIG.provider_fields.get(cls.ENDPOINT, [])
            if isinstance(field, dict)
        )
        + tuple(
            (f"_{CONFIG.provider.prefix}_{field}", field)
            if not field.startswith("_")
            else (field, field)
            for field in cls.PROVIDER_FIELDS
        )
        + tuple(CONFIG.aliases.get(cls.ENDPOINT, {}).items())
        + cls.ALIASES
    )

all_length_aliases() cached classmethod

Returns all of the associated length aliases for this class, including those defined by the server config.

Returns:

Type Description
tuple[tuple[str, str], ...]

A tuple of length alias tuples.
Source code in optimade/server/mappers/entries.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
@classmethod
@lru_cache(maxsize=NUM_ENTRY_TYPES)
def all_length_aliases(cls) -> tuple[tuple[str, str], ...]:
    """Returns all of the associated length aliases for this class,
    including those defined by the server config.

    Returns:
        A tuple of length alias tuples.

    """
    from optimade.server.config import CONFIG

    return cls.LENGTH_ALIASES + tuple(
        CONFIG.length_aliases.get(cls.ENDPOINT, {}).items()
    )

deserialize(results) classmethod

Converts the raw database entries for this class into serialized models, mapping the data along the way.

Source code in optimade/server/mappers/entries.py 
367
368
369
370
371
372
373
374
375
376
377
378
@classmethod
def deserialize(
    cls, results: Union[dict, Iterable[dict]]
) -> Union[list[EntryResource], EntryResource]:
    """Converts the raw database entries for this class into serialized models,
    mapping the data along the way.

    """
    if isinstance(results, dict):
        return cls.ENTRY_RESOURCE_CLASS(**cls.map_back(results))

    return [cls.ENTRY_RESOURCE_CLASS(**cls.map_back(doc)) for doc in results]

get_backend_field(optimade_field) cached classmethod

Return the field name configured for the particular underlying database for the passed OPTIMADE field name, that would be used in an API filter.

Aliases are read from all_aliases().

If a dot-separated OPTIMADE field is provided, e.g., species.mass, only the first part will be mapped. This means for an (OPTIMADE, DB) alias of (species, kinds), get_backend_fields("species.mass") will return kinds.mass.

Parameters:

Name Type Description Default
optimade_field str

The OPTIMADE field to attempt to map to the backend-specific field.

 required

Examples:

>>> get_backend_field("chemical_formula_anonymous")
'formula_anon'
>>> get_backend_field("formula_anon")
'formula_anon'
>>> get_backend_field("_exmpl_custom_provider_field")
'custom_provider_field'

Returns:

Type Description
str

The mapped field name to be used in the query to the backend.
Source code in optimade/server/mappers/entries.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
@classmethod
@lru_cache(maxsize=128)
def get_backend_field(cls, optimade_field: str) -> str:
    """Return the field name configured for the particular
    underlying database for the passed OPTIMADE field name, that would
    be used in an API filter.

    Aliases are read from
    [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

    If a dot-separated OPTIMADE field is provided, e.g., `species.mass`, only the first part will be mapped.
    This means for an (OPTIMADE, DB) alias of (`species`, `kinds`), `get_backend_fields("species.mass")`
    will return `kinds.mass`.

    Arguments:
        optimade_field: The OPTIMADE field to attempt to map to the backend-specific field.

    Examples:
        >>> get_backend_field("chemical_formula_anonymous")
        'formula_anon'
        >>> get_backend_field("formula_anon")
        'formula_anon'
        >>> get_backend_field("_exmpl_custom_provider_field")
        'custom_provider_field'

    Returns:
        The mapped field name to be used in the query to the backend.

    """
    split = optimade_field.split(".")
    alias = dict(cls.all_aliases()).get(split[0], None)
    if alias is not None:
        return alias + ("." + ".".join(split[1:]) if len(split) > 1 else "")
    return optimade_field

get_optimade_field(backend_field) cached classmethod

Return the corresponding OPTIMADE field name for the underlying database field, ready to be used to construct the OPTIMADE-compliant JSON response.

Aliases are read from all_aliases().

Parameters:

Name Type Description Default
backend_field str

The backend field to attempt to map to an OPTIMADE field.

 required

Examples:

>>> get_optimade_field("chemical_formula_anonymous")
'chemical_formula_anonymous'
>>> get_optimade_field("formula_anon")
'chemical_formula_anonymous'
>>> get_optimade_field("custom_provider_field")
'_exmpl_custom_provider_field'

Returns:

Type Description
str

The mapped field name to be used in an OPTIMADE-compliant response.
Source code in optimade/server/mappers/entries.py 
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
@classmethod
@lru_cache(maxsize=128)
def get_optimade_field(cls, backend_field: str) -> str:
    """Return the corresponding OPTIMADE field name for the underlying database field,
    ready to be used to construct the OPTIMADE-compliant JSON response.

    Aliases are read from
    [`all_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_aliases].

    Arguments:
        backend_field: The backend field to attempt to map to an OPTIMADE field.

    Examples:
        >>> get_optimade_field("chemical_formula_anonymous")
        'chemical_formula_anonymous'
        >>> get_optimade_field("formula_anon")
        'chemical_formula_anonymous'
        >>> get_optimade_field("custom_provider_field")
        '_exmpl_custom_provider_field'

    Returns:
        The mapped field name to be used in an OPTIMADE-compliant response.

    """
    return {alias: real for real, alias in cls.all_aliases()}.get(
        backend_field, backend_field
    )

get_required_fields() cached classmethod

Get REQUIRED response fields.

Returns:

Type Description
set

REQUIRED response fields.
Source code in optimade/server/mappers/entries.py 
312
313
314
315
316
317
318
319
320
321
@classmethod
@lru_cache(maxsize=NUM_ENTRY_TYPES)
def get_required_fields(cls) -> set:
    """Get REQUIRED response fields.

    Returns:
        REQUIRED response fields.

    """
    return cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS

length_alias_for(field) cached classmethod

Returns the length alias for the particular field, or None if no such alias is found.

Parameters:

Name Type Description Default
field str

OPTIMADE field name.

 required

Returns:

Type Description
Optional[str]

Aliased field as found in all_length_aliases().
Source code in optimade/server/mappers/entries.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
@classmethod
@lru_cache(maxsize=128)
def length_alias_for(cls, field: str) -> Optional[str]:
    """Returns the length alias for the particular field,
    or `None` if no such alias is found.

    Parameters:
        field: OPTIMADE field name.

    Returns:
        Aliased field as found in [`all_length_aliases()`][optimade.server.mappers.entries.BaseResourceMapper.all_length_aliases].

    """
    return dict(cls.all_length_aliases()).get(field, None)

map_back(doc) classmethod

Map properties from MongoDB to OPTIMADE.

Starting from a MongoDB document doc, map the DB fields to the corresponding OPTIMADE fields. Then, the fields are all added to the top-level field "attributes", with the exception of other top-level fields, defined in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS. All fields not in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS + "attributes" will be removed. Finally, the type is given the value of the specified cls.ENDPOINT.

Parameters:

Name Type Description Default
doc dict

A resource object in MongoDB format.

 required

Returns:

Type Description
dict

A resource object in OPTIMADE format.
Source code in optimade/server/mappers/entries.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
@classmethod
def map_back(cls, doc: dict) -> dict:
    """Map properties from MongoDB to OPTIMADE.

    Starting from a MongoDB document `doc`, map the DB fields to the corresponding OPTIMADE fields.
    Then, the fields are all added to the top-level field "attributes",
    with the exception of other top-level fields, defined in `cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS`.
    All fields not in `cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS` + "attributes" will be removed.
    Finally, the `type` is given the value of the specified `cls.ENDPOINT`.

    Parameters:
        doc: A resource object in MongoDB format.

    Returns:
        A resource object in OPTIMADE format.

    """
    mapping = ((real, alias) for alias, real in cls.all_aliases())
    newdoc = {}
    reals = {real for _, real in cls.all_aliases()}
    for key in doc:
        if key not in reals:
            newdoc[key] = doc[key]
    for real, alias in mapping:
        if real in doc:
            newdoc[alias] = doc[real]

    if "attributes" in newdoc:
        raise Exception("Will overwrite doc field!")
    attributes = newdoc.copy()

    for field in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS:
        value = attributes.pop(field, None)
        if value is not None:
            newdoc[field] = value
    for field in list(newdoc.keys()):
        if field not in cls.TOP_LEVEL_NON_ATTRIBUTES_FIELDS:
            del newdoc[field]

    newdoc["type"] = cls.ENDPOINT
    newdoc["attributes"] = attributes

    return newdoc

EntryCollection

Bases: ABC

Backend-agnostic base class for querying collections of EntryResources.

Source code in optimade/server/entry_collections/entry_collections.py 
 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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
class EntryCollection(ABC):
    """Backend-agnostic base class for querying collections of
    [`EntryResource`][optimade.models.entries.EntryResource]s."""

    pagination_mechanism = PaginationMechanism("page_offset")
    """The default pagination mechansim to use with a given collection,
    if the user does not provide any pagination query parameters.
    """

    def __init__(
        self,
        resource_cls: type[EntryResource],
        resource_mapper: type[BaseResourceMapper],
        transformer: Transformer,
    ):
        """Initialize the collection for the given parameters.

        Parameters:
            resource_cls (EntryResource): The `EntryResource` model
                that is stored by the collection.
            resource_mapper (BaseResourceMapper): A resource mapper
                object that handles aliases and format changes between
                deserialization and response.
            transformer (Transformer): The Lark `Transformer` used to
                interpret the filter.

        """
        self.parser = LarkParser()
        self.resource_cls = resource_cls
        self.resource_mapper = resource_mapper
        self.transformer = transformer

        self.provider_prefix = CONFIG.provider.prefix
        self.provider_fields = [
            field if isinstance(field, str) else field["name"]
            for field in CONFIG.provider_fields.get(resource_mapper.ENDPOINT, [])
        ]

        self._all_fields: set[str] = set()

    @abstractmethod
    def __len__(self) -> int:
        """Returns the total number of entries in the collection."""

    @abstractmethod
    def insert(self, data: list[EntryResource]) -> None:
        """Add the given entries to the underlying database.

        Arguments:
            data: The entry resource objects to add to the database.

        """

    @abstractmethod
    def count(self, **kwargs: Any) -> Optional[int]:
        """Returns the number of entries matching the query specified
        by the keyword arguments.

        Parameters:
            **kwargs: Query parameters as keyword arguments.

        """

    def find(
        self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
    ) -> tuple[
        Optional[Union[dict[str, Any], list[dict[str, Any]]]],
        Optional[int],
        bool,
        set[str],
        set[str],
    ]:
        """
        Fetches results and indicates if more data is available.

        Also gives the total number of data available in the absence of `page_limit`.
        See [`EntryListingQueryParams`][optimade.server.query_params.EntryListingQueryParams]
        for more information.

        Returns a list of the mapped database reponse.

        If no results match the query, then `results` is set to `None`.

        Parameters:
            params: Entry listing URL query params.

        Returns:
            A tuple of various relevant values:
            (`results`, `data_returned`, `more_data_available`, `exclude_fields`, `include_fields`).

        """
        criteria = self.handle_query_params(params)
        single_entry = isinstance(params, SingleEntryQueryParams)
        response_fields: set[str] = criteria.pop("fields")

        raw_results, data_returned, more_data_available = self._run_db_query(
            criteria, single_entry
        )

        exclude_fields = self.all_fields - response_fields
        include_fields = (
            response_fields - self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS
        )

        bad_optimade_fields: set[str] = set()
        bad_provider_fields: set[str] = set()
        supported_prefixes = self.resource_mapper.SUPPORTED_PREFIXES
        all_attributes: set[str] = self.resource_mapper.ALL_ATTRIBUTES
        for field in include_fields:
            if field not in all_attributes:
                if field.startswith("_"):
                    if any(
                        field.startswith(f"_{prefix}_") for prefix in supported_prefixes
                    ):
                        bad_provider_fields.add(field)
                else:
                    bad_optimade_fields.add(field)

        if bad_provider_fields:
            warnings.warn(
                message=f"Unrecognised field(s) for this provider requested in `response_fields`: {bad_provider_fields}.",
                category=UnknownProviderProperty,
            )

        if bad_optimade_fields:
            raise BadRequest(
                detail=f"Unrecognised OPTIMADE field(s) in requested `response_fields`: {bad_optimade_fields}."
            )

        results: Optional[Union[list[dict[str, Any]], dict[str, Any]]] = None

        if raw_results:
            results = [self.resource_mapper.map_back(doc) for doc in raw_results]

            if single_entry:
                results = results[0]

                if (
                    CONFIG.validate_api_response
                    and data_returned is not None
                    and data_returned > 1
                ):
                    raise NotFound(
                        detail=f"Instead of a single entry, {data_returned} entries were found",
                    )
                else:
                    data_returned = 1

        return (
            results,
            data_returned,
            more_data_available,
            exclude_fields,
            include_fields,
        )

    @abstractmethod
    def _run_db_query(
        self, criteria: dict[str, Any], single_entry: bool = False
    ) -> tuple[list[dict[str, Any]], Optional[int], bool]:
        """Run the query on the backend and collect the results.

        Arguments:
            criteria: A dictionary representation of the query parameters.
            single_entry: Whether or not the caller is expecting a single entry response.

        Returns:
            The list of entries from the database (without any re-mapping), the total number of
            entries matching the query and a boolean for whether or not there is more data available.

        """

    @property
    def all_fields(self) -> set[str]:
        """Get the set of all fields handled in this collection,
        from attribute fields in the schema, provider fields and top-level OPTIMADE fields.

        The set of all fields are lazily created and then cached.
        This means the set is created the first time the property is requested and then cached.

        Returns:
            All fields handled in this collection.

        """
        if not self._all_fields:
            # All OPTIMADE fields
            self._all_fields = (
                self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS.copy()
            )
            self._all_fields |= self.get_attribute_fields()
            # All provider-specific fields
            self._all_fields |= {
                f"_{self.provider_prefix}_{field_name}"
                if not field_name.startswith("_")
                else field_name
                for field_name in self.provider_fields
            }

        return self._all_fields

    def get_attribute_fields(self) -> set[str]:
        """Get the set of attribute fields

        Return only the _first-level_ attribute fields from the schema of the resource class,
        resolving references along the way if needed.

        Note:
            It is not needed to take care of other special OpenAPI schema keys than `allOf`,
            since only `allOf` will be found in this context.
            Other special keys can be found in [the Swagger documentation](https://swagger.io/docs/specification/data-models/oneof-anyof-allof-not/).

        Returns:
            Property names.

        """
        annotation = _get_origin_type(
            self.resource_cls.model_fields["attributes"].annotation
        )

        if annotation in (None, NoneType) or not issubclass(annotation, Attributes):
            raise TypeError(
                "resource class 'attributes' field must be a subclass of 'EntryResourceAttributes'"
            )

        return set(annotation.model_fields)  # type: ignore[attr-defined]

    def handle_query_params(
        self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
    ) -> dict[str, Any]:
        """Parse and interpret the backend-agnostic query parameter models into a dictionary
        that can be used by the specific backend.

        Note:
            Currently this method returns the pymongo interpretation of the parameters,
            which will need modification for modified for other backends.

        Parameters:
            params: The initialized query parameter model from the server.

        Raises:
            Forbidden: If too large of a page limit is provided.
            BadRequest: If an invalid request is made, e.g., with incorrect fields
                or response format.

        Returns:
            A dictionary representation of the query parameters.

        """
        cursor_kwargs = {}

        # filter
        if getattr(params, "filter", False):
            cursor_kwargs["filter"] = self.transformer.transform(
                self.parser.parse(params.filter)  # type: ignore[union-attr]
            )
        else:
            cursor_kwargs["filter"] = {}

        # response_format
        if (
            getattr(params, "response_format", False)
            and params.response_format != "json"
        ):
            raise BadRequest(
                detail=f"Response format {params.response_format} is not supported, please use response_format='json'"
            )

        # page_limit
        if getattr(params, "page_limit", False):
            limit = params.page_limit  # type: ignore[union-attr]
            if limit > CONFIG.page_limit_max:
                raise Forbidden(
                    detail=f"Max allowed page_limit is {CONFIG.page_limit_max}, you requested {limit}",
                )
            cursor_kwargs["limit"] = limit
        else:
            cursor_kwargs["limit"] = CONFIG.page_limit

        # response_fields
        cursor_kwargs["projection"] = {
            f"{self.resource_mapper.get_backend_field(f)}": True
            for f in self.all_fields
        }

        if getattr(params, "response_fields", False):
            response_fields = set(params.response_fields.split(","))
            response_fields |= self.resource_mapper.get_required_fields()
        else:
            response_fields = self.all_fields.copy()

        cursor_kwargs["fields"] = response_fields

        # sort
        if getattr(params, "sort", False):
            cursor_kwargs["sort"] = self.parse_sort_params(params.sort)  # type: ignore[union-attr]

        # warn if multiple pagination keys are present, and only use the first from this list
        received_pagination_option = False
        warn_multiple_keys = False

        if getattr(params, "page_offset", False):
            received_pagination_option = True
            cursor_kwargs["skip"] = params.page_offset  # type: ignore[union-attr]

        if isinstance(getattr(params, "page_number", None), int):
            if received_pagination_option:
                warn_multiple_keys = True
            else:
                received_pagination_option = True
                if params.page_number < 1:  # type: ignore[union-attr]
                    warnings.warn(
                        message=f"'page_number' is 1-based, using 'page_number=1' instead of {params.page_number}",  # type: ignore[union-attr]
                        category=QueryParamNotUsed,
                    )
                    page_number = 1
                else:
                    page_number = params.page_number  # type: ignore[union-attr]
                cursor_kwargs["skip"] = (page_number - 1) * cursor_kwargs["limit"]

        if isinstance(getattr(params, "page_above", None), str):
            if received_pagination_option:
                warn_multiple_keys = True
            else:
                received_pagination_option = True
                cursor_kwargs["page_above"] = params.page_above  # type: ignore[union-attr]

        if warn_multiple_keys:
            warnings.warn(
                message="Multiple pagination keys were provided, only using the first one of 'page_offset', 'page_number' or 'page_above'",
                category=QueryParamNotUsed,
            )

        return cursor_kwargs

    def parse_sort_params(self, sort_params: str) -> Iterable[tuple[str, int]]:
        """Handles any sort parameters passed to the collection,
        resolving aliases and dealing with any invalid fields.

        Raises:
            BadRequest: if an invalid sort is requested.

        Returns:
            A list of tuples containing the aliased field name and
            sort direction encoded as 1 (ascending) or -1 (descending).

        """
        sort_spec: list[tuple[str, int]] = []
        for field in sort_params.split(","):
            sort_dir = 1
            if field.startswith("-"):
                field = field[1:]
                sort_dir = -1
            aliased_field = self.resource_mapper.get_backend_field(field)
            sort_spec.append((aliased_field, sort_dir))

        unknown_fields = [
            field
            for field, _ in sort_spec
            if self.resource_mapper.get_optimade_field(field) not in self.all_fields
        ]

        if unknown_fields:
            error_detail = "Unable to sort on unknown field{} '{}'".format(
                "s" if len(unknown_fields) > 1 else "",
                "', '".join(unknown_fields),
            )

            # If all unknown fields are "other" provider-specific, then only provide a warning
            if all(
                (
                    re.match(r"_[a-z_0-9]+_[a-z_0-9]*", field)
                    and not field.startswith(f"_{self.provider_prefix}_")
                )
                for field in unknown_fields
            ):
                warnings.warn(error_detail, FieldValueNotRecognized)

            # Otherwise, if all fields are unknown, or some fields are unknown and do not
            # have other provider prefixes, then return 400: Bad Request
            else:
                raise BadRequest(detail=error_detail)

        # If at least one valid field has been provided for sorting, then use that
        sort_spec = [
            (field, sort_dir)
            for field, sort_dir in sort_spec
            if field not in unknown_fields
        ]

        return sort_spec

    def get_next_query_params(
        self,
        params: EntryListingQueryParams,
        results: Optional[Union[dict[str, Any], list[dict[str, Any]]]],
    ) -> dict[str, list[str]]:
        """Provides url query pagination parameters that will be used in the next
        link.

        Arguments:
            results: The results produced by find.
            params: The parsed request params produced by handle_query_params.

        Returns:
            A dictionary with the necessary query parameters.

        """
        query: dict[str, list[str]] = dict()
        if isinstance(results, list) and results:
            # If a user passed a particular pagination mechanism, keep using it
            # Otherwise, use the default pagination mechanism of the collection
            pagination_mechanism = PaginationMechanism.OFFSET
            for pagination_key in (
                "page_offset",
                "page_number",
                "page_above",
            ):
                if getattr(params, pagination_key, None) is not None:
                    pagination_mechanism = PaginationMechanism(pagination_key)
                    break

            if pagination_mechanism == PaginationMechanism.OFFSET:
                query["page_offset"] = [
                    str(params.page_offset + len(results))  # type: ignore[list-item]
                ]

        return query

all_fields: set[str] property

Get the set of all fields handled in this collection, from attribute fields in the schema, provider fields and top-level OPTIMADE fields.

The set of all fields are lazily created and then cached. This means the set is created the first time the property is requested and then cached.

Returns:

Type Description
set[str]

All fields handled in this collection.

pagination_mechanism = PaginationMechanism('page_offset') class-attribute instance-attribute

The default pagination mechansim to use with a given collection, if the user does not provide any pagination query parameters.

__init__(resource_cls, resource_mapper, transformer)

Initialize the collection for the given parameters.

Parameters:

Name Type Description Default
resource_cls EntryResource

The EntryResource model that is stored by the collection.

 required
resource_mapper BaseResourceMapper

A resource mapper object that handles aliases and format changes between deserialization and response.

 required
transformer Transformer

The Lark Transformer used to interpret the filter.

 required
Source code in optimade/server/entry_collections/entry_collections.py 
 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
def __init__(
    self,
    resource_cls: type[EntryResource],
    resource_mapper: type[BaseResourceMapper],
    transformer: Transformer,
):
    """Initialize the collection for the given parameters.

    Parameters:
        resource_cls (EntryResource): The `EntryResource` model
            that is stored by the collection.
        resource_mapper (BaseResourceMapper): A resource mapper
            object that handles aliases and format changes between
            deserialization and response.
        transformer (Transformer): The Lark `Transformer` used to
            interpret the filter.

    """
    self.parser = LarkParser()
    self.resource_cls = resource_cls
    self.resource_mapper = resource_mapper
    self.transformer = transformer

    self.provider_prefix = CONFIG.provider.prefix
    self.provider_fields = [
        field if isinstance(field, str) else field["name"]
        for field in CONFIG.provider_fields.get(resource_mapper.ENDPOINT, [])
    ]

    self._all_fields: set[str] = set()

__len__() abstractmethod

Returns the total number of entries in the collection.

Source code in optimade/server/entry_collections/entry_collections.py 
117
118
119
@abstractmethod
def __len__(self) -> int:
    """Returns the total number of entries in the collection."""

count(**kwargs) abstractmethod

Returns the number of entries matching the query specified by the keyword arguments.

Parameters:

Name Type Description Default
**kwargs Any

Query parameters as keyword arguments.

 {}
Source code in optimade/server/entry_collections/entry_collections.py 
130
131
132
133
134
135
136
137
138
@abstractmethod
def count(self, **kwargs: Any) -> Optional[int]:
    """Returns the number of entries matching the query specified
    by the keyword arguments.

    Parameters:
        **kwargs: Query parameters as keyword arguments.

    """

find(params)

Fetches results and indicates if more data is available.

Also gives the total number of data available in the absence of page_limit. See EntryListingQueryParams for more information.

Returns a list of the mapped database reponse.

If no results match the query, then results is set to None.

Parameters:

Name Type Description Default
params Union[EntryListingQueryParams, SingleEntryQueryParams]

Entry listing URL query params.

 required

Returns:

Type Description
Optional[Union[dict[str, Any], list[dict[str, Any]]]]

A tuple of various relevant values:
Optional[int]

(results, data_returned, more_data_available, exclude_fields, include_fields).
Source code in optimade/server/entry_collections/entry_collections.py 
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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def find(
    self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
) -> tuple[
    Optional[Union[dict[str, Any], list[dict[str, Any]]]],
    Optional[int],
    bool,
    set[str],
    set[str],
]:
    """
    Fetches results and indicates if more data is available.

    Also gives the total number of data available in the absence of `page_limit`.
    See [`EntryListingQueryParams`][optimade.server.query_params.EntryListingQueryParams]
    for more information.

    Returns a list of the mapped database reponse.

    If no results match the query, then `results` is set to `None`.

    Parameters:
        params: Entry listing URL query params.

    Returns:
        A tuple of various relevant values:
        (`results`, `data_returned`, `more_data_available`, `exclude_fields`, `include_fields`).

    """
    criteria = self.handle_query_params(params)
    single_entry = isinstance(params, SingleEntryQueryParams)
    response_fields: set[str] = criteria.pop("fields")

    raw_results, data_returned, more_data_available = self._run_db_query(
        criteria, single_entry
    )

    exclude_fields = self.all_fields - response_fields
    include_fields = (
        response_fields - self.resource_mapper.TOP_LEVEL_NON_ATTRIBUTES_FIELDS
    )

    bad_optimade_fields: set[str] = set()
    bad_provider_fields: set[str] = set()
    supported_prefixes = self.resource_mapper.SUPPORTED_PREFIXES
    all_attributes: set[str] = self.resource_mapper.ALL_ATTRIBUTES
    for field in include_fields:
        if field not in all_attributes:
            if field.startswith("_"):
                if any(
                    field.startswith(f"_{prefix}_") for prefix in supported_prefixes
                ):
                    bad_provider_fields.add(field)
            else:
                bad_optimade_fields.add(field)

    if bad_provider_fields:
        warnings.warn(
            message=f"Unrecognised field(s) for this provider requested in `response_fields`: {bad_provider_fields}.",
            category=UnknownProviderProperty,
        )

    if bad_optimade_fields:
        raise BadRequest(
            detail=f"Unrecognised OPTIMADE field(s) in requested `response_fields`: {bad_optimade_fields}."
        )

    results: Optional[Union[list[dict[str, Any]], dict[str, Any]]] = None

    if raw_results:
        results = [self.resource_mapper.map_back(doc) for doc in raw_results]

        if single_entry:
            results = results[0]

            if (
                CONFIG.validate_api_response
                and data_returned is not None
                and data_returned > 1
            ):
                raise NotFound(
                    detail=f"Instead of a single entry, {data_returned} entries were found",
                )
            else:
                data_returned = 1

    return (
        results,
        data_returned,
        more_data_available,
        exclude_fields,
        include_fields,
    )

get_attribute_fields()

Get the set of attribute fields

Return only the first-level attribute fields from the schema of the resource class, resolving references along the way if needed.

Note

It is not needed to take care of other special OpenAPI schema keys than allOf, since only allOf will be found in this context. Other special keys can be found in the Swagger documentation.

Returns:

Type Description
set[str]

Property names.
Source code in optimade/server/entry_collections/entry_collections.py 
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
def get_attribute_fields(self) -> set[str]:
    """Get the set of attribute fields

    Return only the _first-level_ attribute fields from the schema of the resource class,
    resolving references along the way if needed.

    Note:
        It is not needed to take care of other special OpenAPI schema keys than `allOf`,
        since only `allOf` will be found in this context.
        Other special keys can be found in [the Swagger documentation](https://swagger.io/docs/specification/data-models/oneof-anyof-allof-not/).

    Returns:
        Property names.

    """
    annotation = _get_origin_type(
        self.resource_cls.model_fields["attributes"].annotation
    )

    if annotation in (None, NoneType) or not issubclass(annotation, Attributes):
        raise TypeError(
            "resource class 'attributes' field must be a subclass of 'EntryResourceAttributes'"
        )

    return set(annotation.model_fields)  # type: ignore[attr-defined]

get_next_query_params(params, results)

Provides url query pagination parameters that will be used in the next link.

Parameters:

Name Type Description Default
results Optional[Union[dict[str, Any], list[dict[str, Any]]]]

The results produced by find.

 required
params EntryListingQueryParams

The parsed request params produced by handle_query_params.

 required

Returns:

Type Description
dict[str, list[str]]

A dictionary with the necessary query parameters.
Source code in optimade/server/entry_collections/entry_collections.py 
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
def get_next_query_params(
    self,
    params: EntryListingQueryParams,
    results: Optional[Union[dict[str, Any], list[dict[str, Any]]]],
) -> dict[str, list[str]]:
    """Provides url query pagination parameters that will be used in the next
    link.

    Arguments:
        results: The results produced by find.
        params: The parsed request params produced by handle_query_params.

    Returns:
        A dictionary with the necessary query parameters.

    """
    query: dict[str, list[str]] = dict()
    if isinstance(results, list) and results:
        # If a user passed a particular pagination mechanism, keep using it
        # Otherwise, use the default pagination mechanism of the collection
        pagination_mechanism = PaginationMechanism.OFFSET
        for pagination_key in (
            "page_offset",
            "page_number",
            "page_above",
        ):
            if getattr(params, pagination_key, None) is not None:
                pagination_mechanism = PaginationMechanism(pagination_key)
                break

        if pagination_mechanism == PaginationMechanism.OFFSET:
            query["page_offset"] = [
                str(params.page_offset + len(results))  # type: ignore[list-item]
            ]

    return query

handle_query_params(params)

Parse and interpret the backend-agnostic query parameter models into a dictionary that can be used by the specific backend.

Note

Currently this method returns the pymongo interpretation of the parameters, which will need modification for modified for other backends.

Parameters:

Name Type Description Default
params Union[EntryListingQueryParams, SingleEntryQueryParams]

The initialized query parameter model from the server.

 required

Raises:

Type Description
Forbidden

If too large of a page limit is provided.
BadRequest

If an invalid request is made, e.g., with incorrect fields or response format.

Returns:

Type Description
dict[str, Any]

A dictionary representation of the query parameters.
Source code in optimade/server/entry_collections/entry_collections.py 
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
def handle_query_params(
    self, params: Union[EntryListingQueryParams, SingleEntryQueryParams]
) -> dict[str, Any]:
    """Parse and interpret the backend-agnostic query parameter models into a dictionary
    that can be used by the specific backend.

    Note:
        Currently this method returns the pymongo interpretation of the parameters,
        which will need modification for modified for other backends.

    Parameters:
        params: The initialized query parameter model from the server.

    Raises:
        Forbidden: If too large of a page limit is provided.
        BadRequest: If an invalid request is made, e.g., with incorrect fields
            or response format.

    Returns:
        A dictionary representation of the query parameters.

    """
    cursor_kwargs = {}

    # filter
    if getattr(params, "filter", False):
        cursor_kwargs["filter"] = self.transformer.transform(
            self.parser.parse(params.filter)  # type: ignore[union-attr]
        )
    else:
        cursor_kwargs["filter"] = {}

    # response_format
    if (
        getattr(params, "response_format", False)
        and params.response_format != "json"
    ):
        raise BadRequest(
            detail=f"Response format {params.response_format} is not supported, please use response_format='json'"
        )

    # page_limit
    if getattr(params, "page_limit", False):
        limit = params.page_limit  # type: ignore[union-attr]
        if limit > CONFIG.page_limit_max:
            raise Forbidden(
                detail=f"Max allowed page_limit is {CONFIG.page_limit_max}, you requested {limit}",
            )
        cursor_kwargs["limit"] = limit
    else:
        cursor_kwargs["limit"] = CONFIG.page_limit

    # response_fields
    cursor_kwargs["projection"] = {
        f"{self.resource_mapper.get_backend_field(f)}": True
        for f in self.all_fields
    }

    if getattr(params, "response_fields", False):
        response_fields = set(params.response_fields.split(","))
        response_fields |= self.resource_mapper.get_required_fields()
    else:
        response_fields = self.all_fields.copy()

    cursor_kwargs["fields"] = response_fields

    # sort
    if getattr(params, "sort", False):
        cursor_kwargs["sort"] = self.parse_sort_params(params.sort)  # type: ignore[union-attr]

    # warn if multiple pagination keys are present, and only use the first from this list
    received_pagination_option = False
    warn_multiple_keys = False

    if getattr(params, "page_offset", False):
        received_pagination_option = True
        cursor_kwargs["skip"] = params.page_offset  # type: ignore[union-attr]

    if isinstance(getattr(params, "page_number", None), int):
        if received_pagination_option:
            warn_multiple_keys = True
        else:
            received_pagination_option = True
            if params.page_number < 1:  # type: ignore[union-attr]
                warnings.warn(
                    message=f"'page_number' is 1-based, using 'page_number=1' instead of {params.page_number}",  # type: ignore[union-attr]
                    category=QueryParamNotUsed,
                )
                page_number = 1
            else:
                page_number = params.page_number  # type: ignore[union-attr]
            cursor_kwargs["skip"] = (page_number - 1) * cursor_kwargs["limit"]

    if isinstance(getattr(params, "page_above", None), str):
        if received_pagination_option:
            warn_multiple_keys = True
        else:
            received_pagination_option = True
            cursor_kwargs["page_above"] = params.page_above  # type: ignore[union-attr]

    if warn_multiple_keys:
        warnings.warn(
            message="Multiple pagination keys were provided, only using the first one of 'page_offset', 'page_number' or 'page_above'",
            category=QueryParamNotUsed,
        )

    return cursor_kwargs

insert(data) abstractmethod

Add the given entries to the underlying database.

Parameters:

Name Type Description Default
data list[EntryResource]

The entry resource objects to add to the database.

 required
Source code in optimade/server/entry_collections/entry_collections.py 
121
122
123
124
125
126
127
128
@abstractmethod
def insert(self, data: list[EntryResource]) -> None:
    """Add the given entries to the underlying database.

    Arguments:
        data: The entry resource objects to add to the database.

    """

parse_sort_params(sort_params)

Handles any sort parameters passed to the collection, resolving aliases and dealing with any invalid fields.

Raises:

Type Description
BadRequest

if an invalid sort is requested.

Returns:

Type Description
Iterable[tuple[str, int]]

A list of tuples containing the aliased field name and
Iterable[tuple[str, int]]

sort direction encoded as 1 (ascending) or -1 (descending).
Source code in optimade/server/entry_collections/entry_collections.py 
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
def parse_sort_params(self, sort_params: str) -> Iterable[tuple[str, int]]:
    """Handles any sort parameters passed to the collection,
    resolving aliases and dealing with any invalid fields.

    Raises:
        BadRequest: if an invalid sort is requested.

    Returns:
        A list of tuples containing the aliased field name and
        sort direction encoded as 1 (ascending) or -1 (descending).

    """
    sort_spec: list[tuple[str, int]] = []
    for field in sort_params.split(","):
        sort_dir = 1
        if field.startswith("-"):
            field = field[1:]
            sort_dir = -1
        aliased_field = self.resource_mapper.get_backend_field(field)
        sort_spec.append((aliased_field, sort_dir))

    unknown_fields = [
        field
        for field, _ in sort_spec
        if self.resource_mapper.get_optimade_field(field) not in self.all_fields
    ]

    if unknown_fields:
        error_detail = "Unable to sort on unknown field{} '{}'".format(
            "s" if len(unknown_fields) > 1 else "",
            "', '".join(unknown_fields),
        )

        # If all unknown fields are "other" provider-specific, then only provide a warning
        if all(
            (
                re.match(r"_[a-z_0-9]+_[a-z_0-9]*", field)
                and not field.startswith(f"_{self.provider_prefix}_")
            )
            for field in unknown_fields
        ):
            warnings.warn(error_detail, FieldValueNotRecognized)

        # Otherwise, if all fields are unknown, or some fields are unknown and do not
        # have other provider prefixes, then return 400: Bad Request
        else:
            raise BadRequest(detail=error_detail)

    # If at least one valid field has been provided for sorting, then use that
    sort_spec = [
        (field, sort_dir)
        for field, sort_dir in sort_spec
        if field not in unknown_fields
    ]

    return sort_spec

EntryListingQueryParams

Bases: BaseQueryParams

Common query params for all Entry listing endpoints.

Attributes:

Name Type Description
filter str

A filter string, in the format described in section API Filtering Format Specification of the specification.
response_format str

The output format requested (see section Response Format). Defaults to the format string 'json', which specifies the standard output format described in this specification.

Example: http://example.com/v1/structures?response_format=xml
email_address EmailStr

An email address of the user making the request. The email SHOULD be that of a person and not an automatic system.

Example: http://example.com/v1/structures?email_address=user@example.com
response_fields str

A comma-delimited set of fields to be provided in the output. If provided, these fields MUST be returned along with the REQUIRED fields. Other OPTIONAL fields MUST NOT be returned when this parameter is present.

Example: http://example.com/v1/structures?response_fields=last_modified,nsites
sort str

If supporting sortable queries, an implementation MUST use the sort query parameter with format as specified by JSON API 1.0.

An implementation MAY support multiple sort fields for a single query. If it does, it again MUST conform to the JSON API 1.0 specification.

If an implementation supports sorting for an entry listing endpoint, then the /info/<entries> endpoint MUST include, for each field name <fieldname> in its data.properties.<fieldname> response value that can be used for sorting, the key sortable with value true. If a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either leave out the sortable key or set it equal to false for the specific field name. The set of field names, with sortable equal to true are allowed to be used in the "sort fields" list according to its definition in the JSON API 1.0 specification. The field sortable is in addition to each property description and other OPTIONAL fields. An example is shown in the section Entry Listing Info Endpoints.
page_limit int

Sets a numerical limit on the number of entries returned. See JSON API 1.0. The API implementation MUST return no more than the number specified. It MAY return fewer. The database MAY have a maximum limit and not accept larger numbers (in which case an error code -- 403 Forbidden -- MUST be returned). The default limit value is up to the API implementation to decide.

Example: http://example.com/optimade/v1/structures?page_limit=100
page_offset int

RECOMMENDED for use with offset-based pagination: using page_offset and page_limit is RECOMMENDED.

Example: Skip 50 structures and fetch up to 100: /structures?page_offset=50&page_limit=100.
page_number int

RECOMMENDED for use with page-based pagination: using page_number and page_limit is RECOMMENDED. It is RECOMMENDED that the first page has number 1, i.e., that page_number is 1-based.

Example: Fetch page 2 of up to 50 structures per page: /structures?page_number=2&page_limit=50.
page_cursor int

RECOMMENDED for use with cursor-based pagination: using page_cursor and page_limit is RECOMMENDED.
page_above str

RECOMMENDED for use with value-based pagination: using page_above/page_below and page_limit is RECOMMENDED.

Example: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by increasing id, so page_above value refers to an id value): /structures?page_above=4000&page_limit=100.
page_below str

RECOMMENDED for use with value-based pagination: using page_above/page_below and page_limit is RECOMMENDED.
include str

A server MAY implement the JSON API concept of returning compound documents by utilizing the include query parameter as specified by JSON API 1.0.

All related resource objects MUST be returned as part of an array value for the top-level included field, see the section JSON Response Schema: Common Fields.

The value of include MUST be a comma-separated list of "relationship paths", as defined in the JSON API. If relationship paths are not supported, or a server is unable to identify a relationship path a 400 Bad Request response MUST be made.

The default value for include is references. This means references entries MUST always be included under the top-level field included as default, since a server assumes if include is not specified by a client in the request, it is still specified as include=references. Note, if a client explicitly specifies include and leaves out references, references resource objects MUST NOT be included under the top-level field included, as per the definition of included, see section JSON Response Schema: Common Fields.

Note: A query with the parameter include set to the empty string means no related resource objects are to be returned under the top-level field included.
api_hint str

If the client provides the parameter, the value SHOULD have the format vMAJOR or vMAJOR.MINOR, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends api_hint=v1.0 to the query string, the hint provided is for major version 1 and minor version 0.
Source code in optimade/server/query_params.py 
 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
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
class EntryListingQueryParams(BaseQueryParams):
    """
    Common query params for all Entry listing endpoints.

    Attributes:
        filter (str): A filter string, in the format described in section API Filtering Format Specification of the specification.

        response_format (str): The output format requested (see section Response Format).
            Defaults to the format string 'json', which specifies the standard output format described in this specification.

            **Example**: `http://example.com/v1/structures?response_format=xml`

        email_address (EmailStr): An email address of the user making the request.
            The email SHOULD be that of a person and not an automatic system.

            **Example**: `http://example.com/v1/structures?email_address=user@example.com`

        response_fields (str): A comma-delimited set of fields to be provided in the output.
            If provided, these fields MUST be returned along with the REQUIRED fields.
            Other OPTIONAL fields MUST NOT be returned when this parameter is present.

            **Example**: `http://example.com/v1/structures?response_fields=last_modified,nsites`

        sort (str): If supporting sortable queries, an implementation MUST use the `sort` query parameter with format as specified
            by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-sorting).

            An implementation MAY support multiple sort fields for a single query.
            If it does, it again MUST conform to the JSON API 1.0 specification.

            If an implementation supports sorting for an entry listing endpoint, then the `/info/<entries>` endpoint MUST include,
            for each field name `<fieldname>` in its `data.properties.<fieldname>` response value that can be used for sorting,
            the key `sortable` with value `true`.
            If a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either
            leave out the `sortable` key or set it equal to `false` for the specific field name.
            The set of field names, with `sortable` equal to `true` are allowed to be used in the "sort fields" list according to
            its definition in the JSON API 1.0 specification.
            The field `sortable` is in addition to each property description and other OPTIONAL fields.
            An example is shown in the section Entry Listing Info Endpoints.

        page_limit (int): Sets a numerical limit on the number of entries returned.
            See [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-pagination).
            The API implementation MUST return no more than the number specified.
            It MAY return fewer. The database MAY have a maximum limit and not accept larger numbers
            (in which case an error code -- `403 Forbidden` -- MUST be returned).
            The default limit value is up to the API implementation to decide.

            **Example**: `http://example.com/optimade/v1/structures?page_limit=100`

        page_offset (int): RECOMMENDED for use with _offset-based_ pagination: using `page_offset` and `page_limit` is RECOMMENDED.

            **Example**: Skip 50 structures and fetch up to 100: `/structures?page_offset=50&page_limit=100`.

        page_number (int): RECOMMENDED for use with _page-based_ pagination: using `page_number` and `page_limit` is RECOMMENDED.
            It is RECOMMENDED that the first page has number 1, i.e., that `page_number` is 1-based.

            **Example**: Fetch page 2 of up to 50 structures per page: `/structures?page_number=2&page_limit=50`.

        page_cursor (int): RECOMMENDED for use with _cursor-based_ pagination: using `page_cursor` and `page_limit` is RECOMMENDED.

        page_above (str): RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.

            **Example**: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by
            increasing `id`, so `page_above` value refers to an `id` value): `/structures?page_above=4000&page_limit=100`.

        page_below (str): RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.

        include (str): A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents)
            by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).

            All related resource objects MUST be returned as part of an array value for the top-level `included` field,
            see the section JSON Response Schema: Common Fields.

            The value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).
            If relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.

            The **default value** for `include` is `references`. This means `references` entries MUST always be included under the top-level field
            `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.
            Note, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level
            field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.

            **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.

        api_hint (str): If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`,
            where MAJOR is a major version and MINOR is a minor version of the API.
            For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.

    """

    # The reference server implementation only supports offset/number-based pagination
    unsupported_params: list[str] = [
        "page_cursor",
        "page_below",
    ]

    def __init__(
        self,
        *,
        filter: Annotated[
            str,
            Query(  # pylint: disable=redefined-builtin
                description="A filter string, in the format described in section API Filtering Format Specification of the specification.",
            ),
        ] = "",
        response_format: Annotated[
            str,
            Query(
                description="The output format requested (see section Response Format).\nDefaults to the format string 'json', which specifies the standard output format described in this specification.\nExample: `http://example.com/v1/structures?response_format=xml`",
            ),
        ] = "json",
        email_address: Annotated[
            EmailStr,
            Query(
                description="An email address of the user making the request.\nThe email SHOULD be that of a person and not an automatic system.\nExample: `http://example.com/v1/structures?email_address=user@example.com`",
            ),
        ] = "",
        response_fields: Annotated[
            str,
            Query(
                description="A comma-delimited set of fields to be provided in the output.\nIf provided, these fields MUST be returned along with the REQUIRED fields.\nOther OPTIONAL fields MUST NOT be returned when this parameter is present.\nExample: `http://example.com/v1/structures?response_fields=last_modified,nsites`",
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        sort: Annotated[
            str,
            Query(
                description='If supporting sortable queries, an implementation MUST use the `sort` query parameter with format as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-sorting).\n\nAn implementation MAY support multiple sort fields for a single query.\nIf it does, it again MUST conform to the JSON API 1.0 specification.\n\nIf an implementation supports sorting for an entry listing endpoint, then the `/info/<entries>` endpoint MUST include, for each field name `<fieldname>` in its `data.properties.<fieldname>` response value that can be used for sorting, the key `sortable` with value `true`.\nIf a field name under an entry listing endpoint supporting sorting cannot be used for sorting, the server MUST either leave out the `sortable` key or set it equal to `false` for the specific field name.\nThe set of field names, with `sortable` equal to `true` are allowed to be used in the "sort fields" list according to its definition in the JSON API 1.0 specification.\nThe field `sortable` is in addition to each property description and other OPTIONAL fields.\nAn example is shown in the section Entry Listing Info Endpoints.',
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        page_limit: Annotated[
            int,
            Query(
                description="Sets a numerical limit on the number of entries returned.\nSee [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-pagination).\nThe API implementation MUST return no more than the number specified.\nIt MAY return fewer.\nThe database MAY have a maximum limit and not accept larger numbers (in which case an error code -- 403 Forbidden -- MUST be returned).\nThe default limit value is up to the API implementation to decide.\nExample: `http://example.com/optimade/v1/structures?page_limit=100`",
                ge=0,
            ),
        ] = CONFIG.page_limit,
        page_offset: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _offset-based_ pagination: using `page_offset` and `page_limit` is RECOMMENDED.\nExample: Skip 50 structures and fetch up to 100: `/structures?page_offset=50&page_limit=100`.",
                ge=0,
            ),
        ] = 0,
        page_number: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _page-based_ pagination: using `page_number` and `page_limit` is RECOMMENDED.\nIt is RECOMMENDED that the first page has number 1, i.e., that `page_number` is 1-based.\nExample: Fetch page 2 of up to 50 structures per page: `/structures?page_number=2&page_limit=50`.",
                # ge=1,  # This constraint is only 'RECOMMENDED' in the specification, so should not be included here or in the OpenAPI schema
            ),
        ] = None,  # type: ignore[assignment]
        page_cursor: Annotated[
            int,
            Query(
                description="RECOMMENDED for use with _cursor-based_ pagination: using `page_cursor` and `page_limit` is RECOMMENDED.",
                ge=0,
            ),
        ] = 0,
        page_above: Annotated[
            str,
            Query(
                description="RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.\nExample: Fetch up to 100 structures above sort-field value 4000 (in this example, server chooses to fetch results sorted by increasing `id`, so `page_above` value refers to an `id` value): `/structures?page_above=4000&page_limit=100`.",
            ),
        ] = None,  # type: ignore[assignment]
        page_below: Annotated[
            str,
            Query(
                description="RECOMMENDED for use with _value-based_ pagination: using `page_above`/`page_below` and `page_limit` is RECOMMENDED.",
            ),
        ] = None,  # type: ignore[assignment]
        include: Annotated[
            str,
            Query(
                description='A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents) by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).\n\nAll related resource objects MUST be returned as part of an array value for the top-level `included` field, see the section JSON Response Schema: Common Fields.\n\nThe value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).\nIf relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.\n\nThe **default value** for `include` is `references`.\nThis means `references` entries MUST always be included under the top-level field `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.\nNote, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.\n\n> **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.',
            ),
        ] = "references",
        api_hint: Annotated[
            str,
            Query(
                description="If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.",
                pattern=r"(v[0-9]+(\.[0-9]+)?)?",
            ),
        ] = "",
    ):
        self.filter = filter
        self.response_format = response_format
        self.email_address = email_address
        self.response_fields = response_fields
        self.sort = sort
        self.page_limit = page_limit
        self.page_offset = page_offset
        self.page_number = page_number
        self.page_cursor = page_cursor
        self.page_above = page_above
        self.page_below = page_below
        self.include = include
        self.api_hint = api_hint

check_params(query_params)

This method checks whether all the query parameters that are specified in the URL string are implemented in the relevant *QueryParams class.

This method handles four cases:

  • If a query parameter is passed that is not defined in the relevant *QueryParams class, and it is not prefixed with a known provider prefix, then a BadRequest is raised.
  • If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with a known provider prefix, then the parameter is silently ignored
  • If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with an unknown provider prefix, then a UnknownProviderQueryParameter warning is emitted.
  • If a query parameter is passed that is on the unsupported_params list for the inherited class, then a QueryParamNotUsed warning is emitted.

Parameters:

Name Type Description Default
query_params Iterable[str]

An iterable of the request's string query parameters.

 required

Raises:

Type Description
`BadRequest`

if the query parameter was not found in the relevant class, or if it does not have a valid prefix.
Source code in optimade/server/query_params.py 
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
def check_params(self, query_params: Iterable[str]) -> None:
    """This method checks whether all the query parameters that are specified
    in the URL string are implemented in the relevant `*QueryParams` class.

    This method handles four cases:

    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      and it is not prefixed with a known provider prefix, then a `BadRequest` is raised.
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with a known provider prefix, then the parameter is silently ignored
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with an unknown provider prefix, then a `UnknownProviderQueryParameter`
      warning is emitted.
    * If a query parameter is passed that is on the `unsupported_params` list for the inherited
      class, then a `QueryParamNotUsed` warning is emitted.

    Arguments:
        query_params: An iterable of the request's string query parameters.

    Raises:
        `BadRequest`: if the query parameter was not found in the relevant class, or if it
            does not have a valid prefix.

    """
    if not getattr(CONFIG, "validate_query_parameters", False):
        return
    errors = []
    warnings = []
    unsupported_warnings = []
    for param in query_params:
        if param in self.unsupported_params:
            unsupported_warnings.append(param)
        if not hasattr(self, param):
            split_param = param.split("_")
            if param.startswith("_") and len(split_param) > 2:
                prefix = split_param[1]
                if prefix in BaseResourceMapper.SUPPORTED_PREFIXES:
                    errors.append(param)
                elif prefix not in BaseResourceMapper.KNOWN_PROVIDER_PREFIXES:
                    warnings.append(param)
            else:
                errors.append(param)

    if warnings:
        warn(
            f"The query parameter(s) '{warnings}' are unrecognised and have been ignored.",
            UnknownProviderQueryParameter,
        )

    if unsupported_warnings:
        warn(
            f"The query parameter(s) '{unsupported_warnings}' are not supported by this server and have been ignored.",
            QueryParamNotUsed,
        )

    if errors:
        raise BadRequest(
            f"The query parameter(s) '{errors}' are not recognised by this endpoint."
        )

EntryResource

Bases: Resource

The base model for an entry resource.

Source code in optimade/models/entries.py 
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
class EntryResource(Resource):
    """The base model for an entry resource."""

    id: Annotated[
        str,
        OptimadeField(
            description="""An entry's ID as defined in section Definition of Terms.

- **Type**: string.

- **Requirements/Conventions**:
    - **Support**: MUST be supported by all implementations, MUST NOT be `null`.
    - **Query**: MUST be a queryable property with support for all mandatory filter features.
    - **Response**: REQUIRED in the response.

- **Examples**:
    - `"db/1234567"`
    - `"cod/2000000"`
    - `"cod/2000000@1234567"`
    - `"nomad/L1234567890"`
    - `"42"`""",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ]

    type: Annotated[
        str,
        OptimadeField(
            description="""The name of the type of an entry.

- **Type**: string.

- **Requirements/Conventions**:
    - **Support**: MUST be supported by all implementations, MUST NOT be `null`.
    - **Query**: MUST be a queryable property with support for all mandatory filter features.
    - **Response**: REQUIRED in the response.
    - MUST be an existing entry type.
    - The entry of type `<type>` and ID `<id>` MUST be returned in response to a request for `/<type>/<id>` under the versioned base URL.

- **Example**: `"structures"`""",
            support=SupportLevel.MUST,
            queryable=SupportLevel.MUST,
        ),
    ]

    attributes: Annotated[
        EntryResourceAttributes,
        StrictField(
            description="""A dictionary, containing key-value pairs representing the entry's properties, except for `type` and `id`.
Database-provider-specific properties need to include the database-provider-specific prefix (see section on Database-Provider-Specific Namespace Prefixes).""",
        ),
    ]

    relationships: Annotated[
        Optional[EntryRelationships],
        StrictField(
            description="""A dictionary containing references to other entries according to the description in section Relationships encoded as [JSON API Relationships](https://jsonapi.org/format/1.0/#document-resource-object-relationships).
The OPTIONAL human-readable description of the relationship MAY be provided in the `description` field inside the `meta` dictionary of the JSON API resource identifier object.""",
        ),
    ] = None

FieldValueNotRecognized

Bases: OptimadeWarning

A field or value used in the request is not recognised by this implementation.

Source code in optimade/warnings.py 
54
55
class FieldValueNotRecognized(OptimadeWarning):
    """A field or value used in the request is not recognised by this implementation."""

Forbidden

Bases: OptimadeHTTPException

403 Forbidden

Source code in optimade/exceptions.py 
71
72
73
74
75
class Forbidden(OptimadeHTTPException):
    """403 Forbidden"""

    status_code: int = 403
    title: str = "Forbidden"

LarkParser

This class wraps a versioned OPTIMADE grammar and allows it to be parsed into Lark tree objects.

Source code in optimade/filterparser/lark_parser.py 
 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
class LarkParser:
    """This class wraps a versioned OPTIMADE grammar and allows
    it to be parsed into Lark tree objects.

    """

    def __init__(
        self, version: Optional[tuple[int, int, int]] = None, variant: str = "default"
    ):
        """For a given version and variant, try to load the corresponding grammar.

        Parameters:
            version: The grammar version number to use (e.g., `(1, 0, 1)` for v1.0.1).
            variant: The grammar variant to employ.

        Raises:
            ParserError: If the requested version/variant of the
                grammar does not exist.

        """

        if not version:
            version = max(
                _ for _ in AVAILABLE_PARSERS if AVAILABLE_PARSERS[_].get("default")
            )

        if version not in AVAILABLE_PARSERS:
            raise ParserError(f"Unknown parser grammar version: {version}")

        if variant not in AVAILABLE_PARSERS[version]:
            raise ParserError(f"Unknown variant of the parser: {variant}")

        self.version = version
        self.variant = variant

        with open(AVAILABLE_PARSERS[version][variant]) as f:
            self.lark = Lark(f, maybe_placeholders=False)

        self.tree: Optional[Tree] = None
        self.filter: Optional[str] = None

    def parse(self, filter_: str) -> Tree:
        """Parse a filter string into a `lark.Tree`.

        Parameters:
            filter_: The filter string to parse.

        Raises:
            BadRequest: If the filter cannot be parsed.

        Returns:
            The parsed filter.

        """
        try:
            self.tree = self.lark.parse(filter_)
            self.filter = filter_
            return self.tree
        except Exception as exc:
            raise BadRequest(
                detail=f"Unable to parse filter {filter_}. Lark traceback: \n{exc}"
            ) from exc

    def __repr__(self):
        if isinstance(self.tree, Tree):
            return self.tree.pretty()
        return repr(self.lark)

__init__(version=None, variant='default')

For a given version and variant, try to load the corresponding grammar.

Parameters:

Name Type Description Default
version Optional[tuple[int, int, int]]

The grammar version number to use (e.g., (1, 0, 1) for v1.0.1).

 None
variant str

The grammar variant to employ.

 'default'

Raises:

Type Description
ParserError

If the requested version/variant of the grammar does not exist.
Source code in optimade/filterparser/lark_parser.py 
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
def __init__(
    self, version: Optional[tuple[int, int, int]] = None, variant: str = "default"
):
    """For a given version and variant, try to load the corresponding grammar.

    Parameters:
        version: The grammar version number to use (e.g., `(1, 0, 1)` for v1.0.1).
        variant: The grammar variant to employ.

    Raises:
        ParserError: If the requested version/variant of the
            grammar does not exist.

    """

    if not version:
        version = max(
            _ for _ in AVAILABLE_PARSERS if AVAILABLE_PARSERS[_].get("default")
        )

    if version not in AVAILABLE_PARSERS:
        raise ParserError(f"Unknown parser grammar version: {version}")

    if variant not in AVAILABLE_PARSERS[version]:
        raise ParserError(f"Unknown variant of the parser: {variant}")

    self.version = version
    self.variant = variant

    with open(AVAILABLE_PARSERS[version][variant]) as f:
        self.lark = Lark(f, maybe_placeholders=False)

    self.tree: Optional[Tree] = None
    self.filter: Optional[str] = None

parse(filter_)

Parse a filter string into a lark.Tree.

Parameters:

Name Type Description Default
filter_ str

The filter string to parse.

 required

Raises:

Type Description
BadRequest

If the filter cannot be parsed.

Returns:

Type Description
Tree

The parsed filter.
Source code in optimade/filterparser/lark_parser.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def parse(self, filter_: str) -> Tree:
    """Parse a filter string into a `lark.Tree`.

    Parameters:
        filter_: The filter string to parse.

    Raises:
        BadRequest: If the filter cannot be parsed.

    Returns:
        The parsed filter.

    """
    try:
        self.tree = self.lark.parse(filter_)
        self.filter = filter_
        return self.tree
    except Exception as exc:
        raise BadRequest(
            detail=f"Unable to parse filter {filter_}. Lark traceback: \n{exc}"
        ) from exc

NotFound

Bases: OptimadeHTTPException

404 Not Found

Source code in optimade/exceptions.py 
78
79
80
81
82
class NotFound(OptimadeHTTPException):
    """404 Not Found"""

    status_code: int = 404
    title: str = "Not Found"

PaginationMechanism

Bases: Enum

The supported pagination mechanisms.

Source code in optimade/server/entry_collections/entry_collections.py 
67
68
69
70
71
72
73
74
class PaginationMechanism(enum.Enum):
    """The supported pagination mechanisms."""

    OFFSET = "page_offset"
    NUMBER = "page_number"
    CURSOR = "page_cursor"
    ABOVE = "page_above"
    BELOW = "page_below"

QueryParamNotUsed

Bases: OptimadeWarning

A query parameter is not used in this request.

Source code in optimade/warnings.py 
62
63
class QueryParamNotUsed(OptimadeWarning):
    """A query parameter is not used in this request."""

SingleEntryQueryParams

Bases: BaseQueryParams

Common query params for single entry endpoints.

Attributes:

Name Type Description
response_format str

The output format requested (see section Response Format). Defaults to the format string 'json', which specifies the standard output format described in this specification.

Example: http://example.com/v1/structures?response_format=xml
email_address EmailStr

An email address of the user making the request. The email SHOULD be that of a person and not an automatic system.

Example: http://example.com/v1/structures?email_address=user@example.com
response_fields str

A comma-delimited set of fields to be provided in the output. If provided, these fields MUST be returned along with the REQUIRED fields. Other OPTIONAL fields MUST NOT be returned when this parameter is present.

Example: http://example.com/v1/structures?response_fields=last_modified,nsites
include str

A server MAY implement the JSON API concept of returning compound documents by utilizing the include query parameter as specified by JSON API 1.0.

All related resource objects MUST be returned as part of an array value for the top-level included field, see the section JSON Response Schema: Common Fields.

The value of include MUST be a comma-separated list of "relationship paths", as defined in the JSON API. If relationship paths are not supported, or a server is unable to identify a relationship path a 400 Bad Request response MUST be made.

The default value for include is references. This means references entries MUST always be included under the top-level field included as default, since a server assumes if include is not specified by a client in the request, it is still specified as include=references. Note, if a client explicitly specifies include and leaves out references, references resource objects MUST NOT be included under the top-level field included, as per the definition of included, see section JSON Response Schema: Common Fields.

Note: A query with the parameter include set to the empty string means no related resource objects are to be returned under the top-level field included.
api_hint str

If the client provides the parameter, the value SHOULD have the format vMAJOR or vMAJOR.MINOR, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends api_hint=v1.0 to the query string, the hint provided is for major version 1 and minor version 0.
Source code in optimade/server/query_params.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
class SingleEntryQueryParams(BaseQueryParams):
    """
    Common query params for single entry endpoints.

    Attributes:
        response_format (str): The output format requested (see section Response Format).
            Defaults to the format string 'json', which specifies the standard output format described in this specification.

            **Example**: `http://example.com/v1/structures?response_format=xml`

        email_address (EmailStr): An email address of the user making the request.
            The email SHOULD be that of a person and not an automatic system.

            **Example**: `http://example.com/v1/structures?email_address=user@example.com`

        response_fields (str): A comma-delimited set of fields to be provided in the output.
            If provided, these fields MUST be returned along with the REQUIRED fields.
            Other OPTIONAL fields MUST NOT be returned when this parameter is present.

            **Example**: `http://example.com/v1/structures?response_fields=last_modified,nsites`

        include (str): A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents)
            by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).

            All related resource objects MUST be returned as part of an array value for the top-level `included` field,
            see the section JSON Response Schema: Common Fields.

            The value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).
            If relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.

            The **default value** for `include` is `references`. This means `references` entries MUST always be included under the top-level field
            `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.
            Note, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level
            field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.

            **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.

        api_hint (str): If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`,
            where MAJOR is a major version and MINOR is a minor version of the API.
            For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.

    """

    def __init__(
        self,
        *,
        response_format: Annotated[
            str,
            Query(
                description="The output format requested (see section Response Format).\nDefaults to the format string 'json', which specifies the standard output format described in this specification.\nExample: `http://example.com/v1/structures?response_format=xml`",
            ),
        ] = "json",
        email_address: Annotated[
            EmailStr,
            Query(
                description="An email address of the user making the request.\nThe email SHOULD be that of a person and not an automatic system.\nExample: `http://example.com/v1/structures?email_address=user@example.com`",
            ),
        ] = "",
        response_fields: Annotated[
            str,
            Query(
                description="A comma-delimited set of fields to be provided in the output.\nIf provided, these fields MUST be returned along with the REQUIRED fields.\nOther OPTIONAL fields MUST NOT be returned when this parameter is present.\nExample: `http://example.com/v1/structures?response_fields=last_modified,nsites`",
                pattern=r"([a-z_][a-z_0-9]*(,[a-z_][a-z_0-9]*)*)?",
            ),
        ] = "",
        include: Annotated[
            str,
            Query(
                description='A server MAY implement the JSON API concept of returning [compound documents](https://jsonapi.org/format/1.0/#document-compound-documents) by utilizing the `include` query parameter as specified by [JSON API 1.0](https://jsonapi.org/format/1.0/#fetching-includes).\n\nAll related resource objects MUST be returned as part of an array value for the top-level `included` field, see the section JSON Response Schema: Common Fields.\n\nThe value of `include` MUST be a comma-separated list of "relationship paths", as defined in the [JSON API](https://jsonapi.org/format/1.0/#fetching-includes).\nIf relationship paths are not supported, or a server is unable to identify a relationship path a `400 Bad Request` response MUST be made.\n\nThe **default value** for `include` is `references`.\nThis means `references` entries MUST always be included under the top-level field `included` as default, since a server assumes if `include` is not specified by a client in the request, it is still specified as `include=references`.\nNote, if a client explicitly specifies `include` and leaves out `references`, `references` resource objects MUST NOT be included under the top-level field `included`, as per the definition of `included`, see section JSON Response Schema: Common Fields.\n\n> **Note**: A query with the parameter `include` set to the empty string means no related resource objects are to be returned under the top-level field `included`.',
            ),
        ] = "references",
        api_hint: Annotated[
            str,
            Query(
                description="If the client provides the parameter, the value SHOULD have the format `vMAJOR` or `vMAJOR.MINOR`, where MAJOR is a major version and MINOR is a minor version of the API. For example, if a client appends `api_hint=v1.0` to the query string, the hint provided is for major version 1 and minor version 0.",
                pattern=r"(v[0-9]+(\.[0-9]+)?)?",
            ),
        ] = "",
    ):
        self.response_format = response_format
        self.email_address = email_address
        self.response_fields = response_fields
        self.include = include
        self.api_hint = api_hint

check_params(query_params)

This method checks whether all the query parameters that are specified in the URL string are implemented in the relevant *QueryParams class.

This method handles four cases:

  • If a query parameter is passed that is not defined in the relevant *QueryParams class, and it is not prefixed with a known provider prefix, then a BadRequest is raised.
  • If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with a known provider prefix, then the parameter is silently ignored
  • If a query parameter is passed that is not defined in the relevant *QueryParams class, that is prefixed with an unknown provider prefix, then a UnknownProviderQueryParameter warning is emitted.
  • If a query parameter is passed that is on the unsupported_params list for the inherited class, then a QueryParamNotUsed warning is emitted.

Parameters:

Name Type Description Default
query_params Iterable[str]

An iterable of the request's string query parameters.

 required

Raises:

Type Description
`BadRequest`

if the query parameter was not found in the relevant class, or if it does not have a valid prefix.
Source code in optimade/server/query_params.py 
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
def check_params(self, query_params: Iterable[str]) -> None:
    """This method checks whether all the query parameters that are specified
    in the URL string are implemented in the relevant `*QueryParams` class.

    This method handles four cases:

    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      and it is not prefixed with a known provider prefix, then a `BadRequest` is raised.
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with a known provider prefix, then the parameter is silently ignored
    * If a query parameter is passed that is not defined in the relevant `*QueryParams` class,
      that is prefixed with an unknown provider prefix, then a `UnknownProviderQueryParameter`
      warning is emitted.
    * If a query parameter is passed that is on the `unsupported_params` list for the inherited
      class, then a `QueryParamNotUsed` warning is emitted.

    Arguments:
        query_params: An iterable of the request's string query parameters.

    Raises:
        `BadRequest`: if the query parameter was not found in the relevant class, or if it
            does not have a valid prefix.

    """
    if not getattr(CONFIG, "validate_query_parameters", False):
        return
    errors = []
    warnings = []
    unsupported_warnings = []
    for param in query_params:
        if param in self.unsupported_params:
            unsupported_warnings.append(param)
        if not hasattr(self, param):
            split_param = param.split("_")
            if param.startswith("_") and len(split_param) > 2:
                prefix = split_param[1]
                if prefix in BaseResourceMapper.SUPPORTED_PREFIXES:
                    errors.append(param)
                elif prefix not in BaseResourceMapper.KNOWN_PROVIDER_PREFIXES:
                    warnings.append(param)
            else:
                errors.append(param)

    if warnings:
        warn(
            f"The query parameter(s) '{warnings}' are unrecognised and have been ignored.",
            UnknownProviderQueryParameter,
        )

    if unsupported_warnings:
        warn(
            f"The query parameter(s) '{unsupported_warnings}' are not supported by this server and have been ignored.",
            QueryParamNotUsed,
        )

    if errors:
        raise BadRequest(
            f"The query parameter(s) '{errors}' are not recognised by this endpoint."
        )

SupportedBackend

Bases: Enum

Enumeration of supported database backends

  • elastic: Elasticsearch.
  • mongodb: MongoDB.
  • mongomock: Also MongoDB, but instead of using the pymongo driver to connect to a live Mongo database instance, this will use the mongomock driver, creating an in-memory database, which is mainly used for testing.
Source code in optimade/server/config.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
class SupportedBackend(Enum):
    """Enumeration of supported database backends

    - `elastic`: [Elasticsearch](https://www.elastic.co/).
    - `mongodb`: [MongoDB](https://www.mongodb.com/).
    - `mongomock`: Also MongoDB, but instead of using the
        [`pymongo`](https://pymongo.readthedocs.io/) driver to connect to a live Mongo
        database instance, this will use the
        [`mongomock`](https://github.com/mongomock/mongomock) driver, creating an
        in-memory database, which is mainly used for testing.

    """

    ELASTIC = "elastic"
    MONGODB = "mongodb"
    MONGOMOCK = "mongomock"

UnknownProviderProperty

Bases: OptimadeWarning

A provider-specific property has been requested via response_fields or as in a filter that is not recognised by this implementation.

Source code in optimade/warnings.py 
78
79
80
81
82
class UnknownProviderProperty(OptimadeWarning):
    """A provider-specific property has been requested via `response_fields` or as in a `filter` that is not
    recognised by this implementation.

    """

_get_origin_type(annotation)

Get the origin type of a type annotation.

Parameters:

Name Type Description Default
annotation type

The type annotation.

 required

Returns:

Type Description
type

The origin type.
Source code in optimade/models/types.py 
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
def _get_origin_type(annotation: type) -> type:
    """Get the origin type of a type annotation.

    Parameters:
        annotation: The type annotation.

    Returns:
        The origin type.

    """
    # If the annotation is a Union, get the first non-None type (this includes
    # Optional[T])
    if isinstance(annotation, (OptionalType, UnionType)):
        for arg in get_args(annotation):
            if arg not in (None, NoneType):
                annotation = arg
                break

    # If the annotation is an Annotated type, get the first type
    if isinstance(annotation, AnnotatedType):
        annotation = get_args(annotation)[0]

    # Recursively unpack annotation, if it is a Union, Optional, or Annotated type
    while isinstance(annotation, (OptionalType, UnionType, AnnotatedType)):
        annotation = _get_origin_type(annotation)

    # Special case for Literal
    # NOTE: Expecting Literal arguments to all be of a single type
    arg = get_args(annotation)
    if arg and not isinstance(arg, type):
        # Expect arg to be a Literal type argument
        annotation = type(arg)

    # Ensure that the annotation is a builtin type
    return getattr(annotation, "__origin__", annotation)

create_collection(name, resource_cls, resource_mapper)

Create an EntryCollection of the configured type, depending on the value of CONFIG.database_backend.

Parameters:

Name Type Description Default
name str

The collection name.

 required
resource_cls type[EntryResource]

The type of entry resource to be stored within the collection.

 required
resource_mapper type[BaseResourceMapper]

The associated resource mapper for that entry resource type.

 required

Returns:

Type Description
EntryCollection

The created EntryCollection.
Source code in optimade/server/entry_collections/entry_collections.py 
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
52
53
54
55
56
57
58
59
60
61
62
63
64
def create_collection(
    name: str,
    resource_cls: type[EntryResource],
    resource_mapper: type[BaseResourceMapper],
) -> "EntryCollection":
    """Create an `EntryCollection` of the configured type, depending on the value of
    `CONFIG.database_backend`.

    Arguments:
        name: The collection name.
        resource_cls: The type of entry resource to be stored within the collection.
        resource_mapper: The associated resource mapper for that entry resource type.

    Returns:
        The created `EntryCollection`.

    """
    if CONFIG.database_backend in (
        SupportedBackend.MONGODB,
        SupportedBackend.MONGOMOCK,
    ):
        from optimade.server.entry_collections.mongo import MongoCollection

        return MongoCollection(
            name=name,
            resource_cls=resource_cls,
            resource_mapper=resource_mapper,
        )

    if CONFIG.database_backend is SupportedBackend.ELASTIC:
        from optimade.server.entry_collections.elasticsearch import ElasticCollection

        return ElasticCollection(
            name=name,
            resource_cls=resource_cls,
            resource_mapper=resource_mapper,
        )

    raise NotImplementedError(
        f"The database backend {CONFIG.database_backend!r} is not implemented"
    )