Module astrapy.operations

# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import annotations

from abc import ABC, abstractmethod
from dataclasses import dataclass
from functools import reduce
from typing import Any, Dict, Iterable, List, Optional

from astrapy.collection import AsyncCollection, Collection
from astrapy.constants import DocumentType, SortType, VectorType
from astrapy.meta import check_deprecated_vector_ize
from astrapy.results import (
    BulkWriteResult,
    DeleteResult,
    InsertManyResult,
    InsertOneResult,
    UpdateResult,
)


def reduce_bulk_write_results(results: List[BulkWriteResult]) -> BulkWriteResult:
    """
    Reduce a list of bulk write results into a single one.

    Args:
        results: a list of BulkWriteResult instances.

    Returns:
        A new BulkWRiteResult object which summarized the whole input list.
    """

    zero = BulkWriteResult.zero()

    def _sum_results(r1: BulkWriteResult, r2: BulkWriteResult) -> BulkWriteResult:
        bulk_api_results = {**r1.bulk_api_results, **r2.bulk_api_results}
        if r1.deleted_count is None or r2.deleted_count is None:
            deleted_count = None
        else:
            deleted_count = r1.deleted_count + r2.deleted_count
        inserted_count = r1.inserted_count + r2.inserted_count
        matched_count = r1.matched_count + r2.matched_count
        modified_count = r1.modified_count + r2.modified_count
        upserted_count = r1.upserted_count + r2.upserted_count
        upserted_ids = {**r1.upserted_ids, **r2.upserted_ids}
        return BulkWriteResult(
            bulk_api_results=bulk_api_results,
            deleted_count=deleted_count,
            inserted_count=inserted_count,
            matched_count=matched_count,
            modified_count=modified_count,
            upserted_count=upserted_count,
            upserted_ids=upserted_ids,
        )

    return reduce(_sum_results, results, zero)


class BaseOperation(ABC):
    """
    Base class for all operations amenable to be used
    in bulk writes on (sync) collections.
    """

    @abstractmethod
    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult: ...


@dataclass
class InsertOne(BaseOperation):
    """
    Represents an `insert_one` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        document: the document to insert.
        vector: an optional suitable vector to enrich the document at insertion.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the document instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the documents instead.
    """

    document: DocumentType
    vector: Optional[VectorType]
    vectorize: Optional[str]

    def __init__(
        self,
        document: DocumentType,
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
    ) -> None:
        self.document = document
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="insert",
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: InsertOneResult = collection.insert_one(
            document=self.document,
            vector=self.vector,
            vectorize=self.vectorize,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class InsertMany(BaseOperation):
    """
    Represents an `insert_many` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        documents: the list document to insert.
        vectors: an optional list of vectors to enrich the documents at insertion.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the documents instead.
        vectorize: an optional list of texts achieving the same effect as `vectors`
            except through an embedding service, if one is configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the documents instead.
        ordered: whether the inserts should be done in sequence.
        chunk_size: how many documents to include in a single API request.
            Exceeding the server maximum allowed value results in an error.
            Leave it unspecified (recommended) to use the system default.
        concurrency: maximum number of concurrent requests to the API at
            a given time. It cannot be more than one for ordered insertions.
    """

    documents: Iterable[DocumentType]
    vectors: Optional[Iterable[Optional[VectorType]]]
    vectorize: Optional[Iterable[Optional[str]]]
    ordered: bool
    chunk_size: Optional[int]
    concurrency: Optional[int]

    def __init__(
        self,
        documents: Iterable[DocumentType],
        *,
        vectors: Optional[Iterable[Optional[VectorType]]] = None,
        vectorize: Optional[Iterable[Optional[str]]] = None,
        ordered: bool = True,
        chunk_size: Optional[int] = None,
        concurrency: Optional[int] = None,
    ) -> None:
        self.documents = documents
        self.ordered = ordered
        check_deprecated_vector_ize(
            vector=None,
            vectors=vectors,
            vectorize=vectorize,
            kind="insert",
            from_operation_class=True,
        )
        self.vectors = vectors
        self.vectorize = vectorize
        self.chunk_size = chunk_size
        self.concurrency = concurrency

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: InsertManyResult = collection.insert_many(
            documents=self.documents,
            vectors=self.vectors,
            vectorize=self.vectorize,
            ordered=self.ordered,
            chunk_size=self.chunk_size,
            concurrency=self.concurrency,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class UpdateOne(BaseOperation):
    """
    Represents an `update_one` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        update: an update prescription to apply to the document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    update: Dict[str, Any]
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        update: Dict[str, Any],
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.update = update
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort
        self.upsert = upsert

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = collection.update_one(
            filter=self.filter,
            update=self.update,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class UpdateMany(BaseOperation):
    """
    Represents an `update_many` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select target documents.
        update: an update prescription to apply to the documents.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    update: Dict[str, Any]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        update: Dict[str, Any],
        *,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.update = update
        self.upsert = upsert

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = collection.update_many(
            filter=self.filter,
            update=self.update,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class ReplaceOne(BaseOperation):
    """
    Represents a `replace_one` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        replacement: the replacement document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    replacement: DocumentType
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        replacement: DocumentType,
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.replacement = replacement
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort
        self.upsert = upsert

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = collection.replace_one(
            filter=self.filter,
            replacement=self.replacement,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class DeleteOne(BaseOperation):
    """
    Represents a `delete_one` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
    """

    filter: Dict[str, Any]
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]

    def __init__(
        self,
        filter: Dict[str, Any],
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
    ) -> None:
        self.filter = filter
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: DeleteResult = collection.delete_one(
            filter=self.filter,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class DeleteMany(BaseOperation):
    """
    Represents a `delete_many` operation on a (sync) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select target documents.
    """

    filter: Dict[str, Any]

    def __init__(
        self,
        filter: Dict[str, Any],
    ) -> None:
        self.filter = filter

    def execute(
        self,
        collection: Collection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: DeleteResult = collection.delete_many(
            filter=self.filter, max_time_ms=bulk_write_timeout_ms
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


class AsyncBaseOperation(ABC):
    """
    Base class for all operations amenable to be used
    in bulk writes on (async) collections.
    """

    @abstractmethod
    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult: ...


@dataclass
class AsyncInsertOne(AsyncBaseOperation):
    """
    Represents an `insert_one` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        document: the document to insert.
        vector: an optional suitable vector to enrich the document at insertion.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the document instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the document instead.
    """

    document: DocumentType
    vector: Optional[VectorType]
    vectorize: Optional[str]

    def __init__(
        self,
        document: DocumentType,
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
    ) -> None:
        self.document = document
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="insert",
            from_async_method=True,
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: InsertOneResult = await collection.insert_one(
            document=self.document,
            vector=self.vector,
            vectorize=self.vectorize,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncInsertMany(AsyncBaseOperation):
    """
    Represents an `insert_many` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        documents: the list document to insert.
        vectors: an optional list of vectors to enrich the documents at insertion.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the documents instead.
        vectorize: an optional list of texts achieving the same effect as `vectors`
            except through an embedding service, if one is configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the documents instead.
        ordered: whether the inserts should be done in sequence.
        chunk_size: how many documents to include in a single API request.
            Exceeding the server maximum allowed value results in an error.
            Leave it unspecified (recommended) to use the system default.
        concurrency: maximum number of concurrent requests to the API at
            a given time. It cannot be more than one for ordered insertions.
    """

    documents: Iterable[DocumentType]
    vectors: Optional[Iterable[Optional[VectorType]]]
    vectorize: Optional[Iterable[Optional[str]]]
    ordered: bool
    chunk_size: Optional[int]
    concurrency: Optional[int]

    def __init__(
        self,
        documents: Iterable[DocumentType],
        *,
        vectors: Optional[Iterable[Optional[VectorType]]] = None,
        vectorize: Optional[Iterable[Optional[str]]] = None,
        ordered: bool = True,
        chunk_size: Optional[int] = None,
        concurrency: Optional[int] = None,
    ) -> None:
        self.documents = documents
        check_deprecated_vector_ize(
            vector=None,
            vectors=vectors,
            vectorize=vectorize,
            kind="insert",
            from_async_method=True,
            from_operation_class=True,
        )
        self.vectors = vectors
        self.vectorize = vectorize
        self.ordered = ordered
        self.chunk_size = chunk_size
        self.concurrency = concurrency

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: InsertManyResult = await collection.insert_many(
            documents=self.documents,
            vectors=self.vectors,
            vectorize=self.vectorize,
            ordered=self.ordered,
            chunk_size=self.chunk_size,
            concurrency=self.concurrency,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncUpdateOne(AsyncBaseOperation):
    """
    Represents an `update_one` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        update: an update prescription to apply to the document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    update: Dict[str, Any]
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        update: Dict[str, Any],
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.update = update
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_async_method=True,
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort
        self.upsert = upsert

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = await collection.update_one(
            filter=self.filter,
            update=self.update,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncUpdateMany(AsyncBaseOperation):
    """
    Represents an `update_many` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select target documents.
        update: an update prescription to apply to the documents.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    update: Dict[str, Any]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        update: Dict[str, Any],
        *,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.update = update
        self.upsert = upsert

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = await collection.update_many(
            filter=self.filter,
            update=self.update,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncReplaceOne(AsyncBaseOperation):
    """
    Represents a `replace_one` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        replacement: the replacement document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
        upsert: controls what to do when no documents are found.
    """

    filter: Dict[str, Any]
    replacement: DocumentType
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]
    upsert: bool

    def __init__(
        self,
        filter: Dict[str, Any],
        replacement: DocumentType,
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
        upsert: bool = False,
    ) -> None:
        self.filter = filter
        self.replacement = replacement
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_async_method=True,
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort
        self.upsert = upsert

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: UpdateResult = await collection.replace_one(
            filter=self.filter,
            replacement=self.replacement,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            upsert=self.upsert,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncDeleteOne(AsyncBaseOperation):
    """
    Represents a `delete_one` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select a target document.
        vector: a vector of numbers to use for ANN (vector-search) sorting.
            *DEPRECATED* (removal in 2.0). Use a `$vector` key in the
            sort clause dict instead.
        vectorize: a string to be made into a vector, with the same result as the
            `vector` attribute, through an embedding service, assuming one is
            configured for the collection.
            *DEPRECATED* (removal in 2.0). Use a `$vectorize` key in the
            sort clause dict instead.
        sort: controls ordering of results, hence which document is affected.
    """

    filter: Dict[str, Any]
    vector: Optional[VectorType]
    vectorize: Optional[str]
    sort: Optional[SortType]

    def __init__(
        self,
        filter: Dict[str, Any],
        *,
        vector: Optional[VectorType] = None,
        vectorize: Optional[str] = None,
        sort: Optional[SortType] = None,
    ) -> None:
        self.filter = filter
        check_deprecated_vector_ize(
            vector=vector,
            vectors=None,
            vectorize=vectorize,
            kind="find",
            from_async_method=True,
            from_operation_class=True,
        )
        self.vector = vector
        self.vectorize = vectorize
        self.sort = sort

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: DeleteResult = await collection.delete_one(
            filter=self.filter,
            vector=self.vector,
            vectorize=self.vectorize,
            sort=self.sort,
            max_time_ms=bulk_write_timeout_ms,
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)


@dataclass
class AsyncDeleteMany(AsyncBaseOperation):
    """
    Represents a `delete_many` operation on a (async) collection.
    See the documentation on the collection method for more information.

    Attributes:
        filter: a filter condition to select target documents.
    """

    filter: Dict[str, Any]

    def __init__(
        self,
        filter: Dict[str, Any],
    ) -> None:
        self.filter = filter

    async def execute(
        self,
        collection: AsyncCollection,
        index_in_bulk_write: int,
        bulk_write_timeout_ms: Optional[int],
    ) -> BulkWriteResult:
        """
        Execute this operation against a collection as part of a bulk write.

        Args:
            collection: the collection this write targets.
            insert_in_bulk_write: the index in the list of bulkoperations
        """

        op_result: DeleteResult = await collection.delete_many(
            filter=self.filter, max_time_ms=bulk_write_timeout_ms
        )
        return op_result.to_bulk_write_result(index_in_bulk_write=index_in_bulk_write)