def _aggregate_one_result(
            self, sock_info, slave_ok, cmd, collation=None, session=None):
        """Internal helper to run an aggregate that returns a single result."""
        result = self._command(
            sock_info,
            cmd,
            slave_ok,
            codec_options=self.__write_response_codec_options,
            read_concern=self.read_concern,
            collation=collation,
            session=session)
        batch = result['cursor']['firstBatch']
        return batch[0] if batch else None
Internal helper to run an aggregate that returns a single result.