Skip to content

utils

Utility functions for the queries module.

update_query(query, field, value, operator=None, **mongo_kwargs) async

Update a query's field attribute with value.

If field is a dot-separated value, then only the last field part may be a non-pre-existing field. Otherwise a KeyError or AttributeError will be raised.

Note

This can only update a field for a query's attributes, i.e., this function cannot update id, type or any other top-level resource field.

Important

mongo_kwargs will not be considered for updating the pydantic model instance.

Parameters:

Name Type Description Default
query QueryResource

The query to be updated.

 required
field str

The attributes field (key) to be set. This can be a dot-separated key value to signify embedded fields.

Example: response.meta.

 required
value Any

The (possibly) new value for field.

 required
operator str | None

A MongoDB operator to be used for updating field with value.

 None
**mongo_kwargs Any

Further MongoDB update filters.

 {}
Source code in optimade_gateway/queries/utils.py 
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 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
async def update_query(
    query: QueryResource,
    field: str,
    value: Any,
    operator: str | None = None,
    **mongo_kwargs: Any,
) -> None:
    """Update a query's `field` attribute with `value`.

    If `field` is a dot-separated value, then only the last field part may be a
    non-pre-existing field. Otherwise a `KeyError` or `AttributeError` will be raised.

    !!! note
        This can *only* update a field for a query's `attributes`, i.e., this function
        cannot update `id`, `type` or any other top-level resource field.

    !!! important
        `mongo_kwargs` will not be considered for updating the pydantic model instance.

    Parameters:
        query: The query to be updated.
        field: The `attributes` field (key) to be set.
            This can be a dot-separated key value to signify embedded fields.

            **Example**: `response.meta`.
        value: The (possibly) new value for `field`.
        operator: A MongoDB operator to be used for updating `field` with `value`.
        **mongo_kwargs: Further MongoDB update filters.

    """
    operator = operator or "$set"

    if operator and not operator.startswith("$"):
        operator = f"${operator}"

    update_time = datetime.now(timezone.utc)

    update_kwargs = {"$set": {"last_modified": update_time}}

    if mongo_kwargs:
        update_kwargs.update(mongo_kwargs)

    if operator and operator == "$set":
        update_kwargs["$set"].update({field: value})
    elif operator:
        if operator in update_kwargs:
            update_kwargs[operator].update({field: value})
        else:
            update_kwargs.update({operator: {field: value}})

    # MongoDB
    collection = await collection_factory(CONFIG.queries_collection)
    result: UpdateResult = await collection.collection.update_one(
        filter={"id": {"$eq": query.id}},
        update=await clean_python_types(update_kwargs),
    )
    if result.matched_count != 1:
        LOGGER.error(
            (
                "matched_count should have been exactly 1, it was: %s. "
                "Returned update_one result: %s"
            ),
            result.matched_count,
            result.raw_result,
        )

    # Pydantic model instance
    query.attributes.last_modified = update_time
    if "." in field:
        field_list = field.split(".")
        sub_field: BaseModel | dict[str, Any] = getattr(query.attributes, field_list[0])
        for field_part in field_list[1:-1]:
            if isinstance(sub_field, dict):
                sub_field = sub_field.get(field_part, {})
            else:
                sub_field = getattr(sub_field, field_part)
        if isinstance(sub_field, dict):
            sub_field[field_list[-1]] = value
        else:
            setattr(sub_field, field_list[-1], value)
    else:
        setattr(query.attributes, field, value)