Function forward

backend/data_process/tasks.py:1622–1951

Vectorize and store processed chunks in Elasticsearch Args: processed_data: Dict containing chunks and metadata index_name: Name of the index to store documents source: Original source path (for metadata) source_type: The type of the source("local", "minio")

(
        self,
        processed_data: Dict,
        index_name: str,
        source: str,
        source_type: str = 'minio',
        original_filename: Optional[str] = None,
        authorization: Optional[str] = None
)

@app.task(bind=True, base=LoggingTask, name='data_process.tasks.forward', queue='forward_q')
def forward(
        self,
        processed_data: Dict,
        index_name: str,
        source: str,
        source_type: str = 'minio',
        original_filename: Optional[str] = None,
        authorization: Optional[str] = None
) -> Dict:
    """
    Vectorize and store processed chunks in Elasticsearch

    Args:
        processed_data: Dict containing chunks and metadata
        index_name: Name of the index to store documents
        source: Original source path (for metadata)
        source_type: The type of the source("local", "minio")
        original_filename: The original name of the file
        authorization: Authorization header for API calls

    Returns:
        Dict containing storage results and metadata
    """
    start_time = time.time()
    task_id = self.request.id
    # _warn_if_queue_mismatch("FORWARD TASK", "forward_q", self.request)
    original_source = source
    original_index_name = index_name
    filename = original_filename

    try:
        ctx = _init_forward_context(
            task_id=task_id,
            request_id=str(self.request.id),
            start_time=start_time,
            source=source,
            index_name=index_name,
            source_type=source_type,
            original_filename=original_filename,
        )

        # Before doing any heavy work, check whether this task has been explicitly cancelled.
        if _is_forward_task_cancelled(ctx):
            logger.info(
                f"[{self.request.id}] FORWARD TASK: Detected cancellation flag for task {task_id}; "
                f"skipping chunk forwarding for source '{source}' in index '{index_name}'."
            )
            return _build_forward_cancelled_result(ctx)

        chunks, split_async, original_source, original_index_name, filename = _load_forward_chunks(
            self,
            processed_data=processed_data,
            original_source=original_source,
            original_index_name=original_index_name,
            filename=filename,
        )

        # Calculate total chunks for progress tracking