Skip to content

mongo

MongoCollection

Bases: EntryCollection

Class for querying MongoDB collections (implemented by either pymongo or mongomock) containing serialized EntryResources objects.

Source code in optimade/server/entry_collections/mongo.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
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
class MongoCollection(EntryCollection):
    """Class for querying MongoDB collections (implemented by either pymongo or mongomock)
    containing serialized [`EntryResource`][optimade.models.entries.EntryResource]s objects.

    """

    def __init__(
        self,
        name: str,
        resource_cls: type[EntryResource],
        resource_mapper: type[BaseResourceMapper],
        database: str = CONFIG.mongo_database,
    ):
        """Initialize the MongoCollection for the given parameters.

        Parameters:
            name: The name of the collection.
            resource_cls: The type of entry resource that is stored by the collection.
            resource_mapper: A resource mapper object that handles aliases and
                format changes between deserialization and response.
            database: The name of the underlying MongoDB database to connect to.

        """
        super().__init__(
            resource_cls,
            resource_mapper,
            MongoTransformer(mapper=resource_mapper),
        )

        self.collection = CLIENT[database][name]

        # check aliases do not clash with mongo operators
        self._check_aliases(self.resource_mapper.all_aliases())
        self._check_aliases(self.resource_mapper.all_length_aliases())

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

    def count(self, **kwargs: Any) -> int | None:
        """Returns the number of entries matching the query specified
        by the keyword arguments, or `None` if the count timed out.

        Parameters:
            **kwargs: Query parameters as keyword arguments. The keys
                'filter', 'skip', 'limit', 'hint' and 'maxTimeMS' will be passed
                to the `pymongo.collection.Collection.count_documents` method.

        """
        for k in list(kwargs.keys()):
            if k not in ("filter", "skip", "limit", "hint", "maxTimeMS"):
                del kwargs[k]
        if "filter" not in kwargs:
            return self.collection.estimated_document_count()
        else:
            if "maxTimeMS" not in kwargs:
                kwargs["maxTimeMS"] = 1000 * CONFIG.mongo_count_timeout
            try:
                return self.collection.count_documents(**kwargs)
            except ExecutionTimeout:
                return None

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

        Warning:
            No validation is performed on the incoming data, this data
            should have been mapped to the appropriate format before
            insertion.

        Arguments:
            data: The entries to add to the database.

        """
        self.collection.insert_many(data, ordered=False)

    def create_index(self, field: str, unique: bool = False) -> None:
        """Create an index on the given field, as stored in the database.

        If any error is raised during index creation, this method should faithfully
        return it, except for the simple case where an identical index already exists.

        Arguments:
            field: The database field to index (i.e., if different from the OPTIMADE field,
                the mapper should be used to convert between the two).
            unique: Whether or not the index should be unique.

        """
        self.collection.create_index(field, unique=unique, background=True)

    def create_default_index(self) -> None:
        """Create the default index for the collection.

        For MongoDB, the default is to create a unique index
        on the `id` field. This method should obey any configured
        mappers.

        """
        self.create_index(self.resource_mapper.get_backend_field("id"), unique=True)

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

        This Mongo-specific method calls the base `EntryCollection.handle_query_params` method
        and adds additional handling of the MongoDB ObjectID type.

        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.

        """
        criteria = super().handle_query_params(params)
        # Handle MongoDB ObjectIDs:
        # - If they were not requested, then explicitly remove them
        # - If they were requested, then cast them to strings in the response
        if "_id" not in criteria.get("projection", {}):
            criteria["projection"]["_id"] = False

        if "page_above" in criteria:
            raise NotImplementedError(
                "`page_above` is not implemented for this backend."
            )

        if criteria.get("projection", {}).get("_id"):
            criteria["projection"]["_id"] = {"$toString": "$_id"}

        return criteria

    def _run_db_query(
        self, criteria: dict[str, Any], single_entry: bool = False
    ) -> tuple[list[dict[str, Any]], int | None, 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.

        """
        results = list(self.collection.find(**criteria))

        if CONFIG.database_backend == SupportedBackend.MONGOMOCK and criteria.get(
            "projection", {}
        ).get("_id"):
            # mongomock does not support `$toString`` in projection, so we have to do it manually
            for ind, doc in enumerate(results):
                results[ind]["_id"] = str(doc["_id"])

        nresults_now = len(results)
        if not single_entry:
            criteria_nolimit = criteria.copy()
            criteria_nolimit.pop("limit", None)
            skip = criteria_nolimit.pop("skip", 0)
            data_returned = self.count(**criteria_nolimit)
            # Only correct most of the time: if the total number of remaining results is exactly the page limit
            # then this will incorrectly say there is more_data_available
            if data_returned is None:
                more_data_available = nresults_now == criteria.get("limit", 0)
            else:
                more_data_available = nresults_now + skip < data_returned
        else:
            # SingleEntryQueryParams, e.g., /structures/{entry_id}
            data_returned = nresults_now
            more_data_available = False

        return results, data_returned, more_data_available

    def _check_aliases(self, aliases):
        """Check that aliases do not clash with mongo keywords."""
        if any(
            alias[0].startswith("$") or alias[1].startswith("$") for alias in aliases
        ):
            raise RuntimeError(f"Cannot define an alias starting with a '$': {aliases}")

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__(name, resource_cls, resource_mapper, database=CONFIG.mongo_database)

Initialize the MongoCollection for the given parameters.

Parameters:

Name Type Description Default
name str

The name of the collection.

 required
resource_cls type[EntryResource]

The type of entry resource that is stored by the collection.

 required
resource_mapper type[BaseResourceMapper]

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

 required
database str

The name of the underlying MongoDB database to connect to.

 mongo_database
Source code in optimade/server/entry_collections/mongo.py 
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
def __init__(
    self,
    name: str,
    resource_cls: type[EntryResource],
    resource_mapper: type[BaseResourceMapper],
    database: str = CONFIG.mongo_database,
):
    """Initialize the MongoCollection for the given parameters.

    Parameters:
        name: The name of the collection.
        resource_cls: The type of entry resource that is stored by the collection.
        resource_mapper: A resource mapper object that handles aliases and
            format changes between deserialization and response.
        database: The name of the underlying MongoDB database to connect to.

    """
    super().__init__(
        resource_cls,
        resource_mapper,
        MongoTransformer(mapper=resource_mapper),
    )

    self.collection = CLIENT[database][name]

    # check aliases do not clash with mongo operators
    self._check_aliases(self.resource_mapper.all_aliases())
    self._check_aliases(self.resource_mapper.all_length_aliases())

__len__()

Returns the total number of entries in the collection.

Source code in optimade/server/entry_collections/mongo.py 
67
68
69
def __len__(self) -> int:
    """Returns the total number of entries in the collection."""
    return self.collection.estimated_document_count()

count(**kwargs)

Returns the number of entries matching the query specified by the keyword arguments, or None if the count timed out.

Parameters:

Name Type Description Default
**kwargs Any

Query parameters as keyword arguments. The keys 'filter', 'skip', 'limit', 'hint' and 'maxTimeMS' will be passed to the pymongo.collection.Collection.count_documents method.

 {}
Source code in optimade/server/entry_collections/mongo.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def count(self, **kwargs: Any) -> int | None:
    """Returns the number of entries matching the query specified
    by the keyword arguments, or `None` if the count timed out.

    Parameters:
        **kwargs: Query parameters as keyword arguments. The keys
            'filter', 'skip', 'limit', 'hint' and 'maxTimeMS' will be passed
            to the `pymongo.collection.Collection.count_documents` method.

    """
    for k in list(kwargs.keys()):
        if k not in ("filter", "skip", "limit", "hint", "maxTimeMS"):
            del kwargs[k]
    if "filter" not in kwargs:
        return self.collection.estimated_document_count()
    else:
        if "maxTimeMS" not in kwargs:
            kwargs["maxTimeMS"] = 1000 * CONFIG.mongo_count_timeout
        try:
            return self.collection.count_documents(**kwargs)
        except ExecutionTimeout:
            return None

create_default_index()

Create the default index for the collection.

For MongoDB, the default is to create a unique index on the id field. This method should obey any configured mappers.

Source code in optimade/server/entry_collections/mongo.py 
122
123
124
125
126
127
128
129
130
def create_default_index(self) -> None:
    """Create the default index for the collection.

    For MongoDB, the default is to create a unique index
    on the `id` field. This method should obey any configured
    mappers.

    """
    self.create_index(self.resource_mapper.get_backend_field("id"), unique=True)

create_index(field, unique=False)

Create an index on the given field, as stored in the database.

If any error is raised during index creation, this method should faithfully return it, except for the simple case where an identical index already exists.

Parameters:

Name Type Description Default
field str

The database field to index (i.e., if different from the OPTIMADE field, the mapper should be used to convert between the two).

 required
unique bool

Whether or not the index should be unique.

 False
Source code in optimade/server/entry_collections/mongo.py 
108
109
110
111
112
113
114
115
116
117
118
119
120
def create_index(self, field: str, unique: bool = False) -> None:
    """Create an index on the given field, as stored in the database.

    If any error is raised during index creation, this method should faithfully
    return it, except for the simple case where an identical index already exists.

    Arguments:
        field: The database field to index (i.e., if different from the OPTIMADE field,
            the mapper should be used to convert between the two).
        unique: Whether or not the index should be unique.

    """
    self.collection.create_index(field, unique=unique, background=True)

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 EntryListingQueryParams | SingleEntryQueryParams

Entry listing URL query params.

 required

Returns:

Type Description
dict[str, Any] | list[dict[str, Any]] | None

A tuple of various relevant values:
int | None

(results, data_returned, more_data_available, exclude_fields, include_fields).
Source code in optimade/server/entry_collections/entry_collections.py 
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
def find(
    self, params: EntryListingQueryParams | SingleEntryQueryParams
) -> tuple[
    dict[str, Any] | list[dict[str, Any]] | None,
    int | None,
    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: list[dict[str, Any]] | dict[str, Any] | None = 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 
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
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 dict[str, Any] | list[dict[str, Any]] | None

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 
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
def get_next_query_params(
    self,
    params: EntryListingQueryParams,
    results: dict[str, Any] | list[dict[str, Any]] | None,
) -> 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 MongoDB.

This Mongo-specific method calls the base EntryCollection.handle_query_params method and adds additional handling of the MongoDB ObjectID type.

Parameters:

Name Type Description Default
params 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/mongo.py 
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
def handle_query_params(
    self, params: EntryListingQueryParams | SingleEntryQueryParams
) -> dict[str, Any]:
    """Parse and interpret the backend-agnostic query parameter models into a dictionary
    that can be used by MongoDB.

    This Mongo-specific method calls the base `EntryCollection.handle_query_params` method
    and adds additional handling of the MongoDB ObjectID type.

    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.

    """
    criteria = super().handle_query_params(params)
    # Handle MongoDB ObjectIDs:
    # - If they were not requested, then explicitly remove them
    # - If they were requested, then cast them to strings in the response
    if "_id" not in criteria.get("projection", {}):
        criteria["projection"]["_id"] = False

    if "page_above" in criteria:
        raise NotImplementedError(
            "`page_above` is not implemented for this backend."
        )

    if criteria.get("projection", {}).get("_id"):
        criteria["projection"]["_id"] = {"$toString": "$_id"}

    return criteria

insert(data)

Add the given entries to the underlying database.

Warning

No validation is performed on the incoming data, this data should have been mapped to the appropriate format before insertion.

Parameters:

Name Type Description Default
data list[EntryResource | dict]

The entries to add to the database.

 required
Source code in optimade/server/entry_collections/mongo.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def insert(self, data: list[EntryResource | dict]) -> None:
    """Add the given entries to the underlying database.

    Warning:
        No validation is performed on the incoming data, this data
        should have been mapped to the appropriate format before
        insertion.

    Arguments:
        data: The entries to add to the database.

    """
    self.collection.insert_many(data, ordered=False)

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