Method _aggregate

pymongo/synchronous/collection.py:2914–2942

(
        self,
        aggregation_command: Type[_AggregationCommand],
        pipeline: _Pipeline,
        cursor_class: Type[CommandCursor],  # type: ignore[type-arg]
        session: Optional[ClientSession],
        let: Optional[Mapping[str, Any]] = None,
        comment: Optional[Any] = None,
        **kwargs: Any,
    )

    @_csot.apply
    def _aggregate(
        self,
        aggregation_command: Type[_AggregationCommand],
        pipeline: _Pipeline,
        cursor_class: Type[CommandCursor],  # type: ignore[type-arg]
        session: Optional[ClientSession],
        let: Optional[Mapping[str, Any]] = None,
        comment: Optional[Any] = None,
        **kwargs: Any,
    ) -> CommandCursor[_DocumentType]:
        if comment is not None:
            kwargs["comment"] = comment
        cmd = aggregation_command(
            self,
            cursor_class,
            pipeline,
            kwargs,
            let,
            user_fields={"cursor": {"firstBatch": 1}},
        )

        return self._database.client._retryable_read(
            cmd.get_cursor,
            cmd.get_read_preference(session),  # type: ignore[arg-type]
            session,
            retryable=not cmd._performs_write,
            operation=_Op.AGGREGATE,
            is_aggregate_write=cmd._performs_write,
        )

    def aggregate(
        self,