Function process_sync

backend/data_process/tasks.py:2146–2260

Synchronous process task that returns text directly (for real-time API) Args: source: Source file path, URL, or text content source_type: source of the file("local", "minio") chunking_strategy: Strategy for chunking the document timeout: Timeout for the oper

(
        self,
        source: str,
        source_type: str,
        chunking_strategy: str = "basic",
        timeout: int = 30,
        **params
)

@app.task(bind=True, base=LoggingTask, name='data_process.tasks.process_sync')
def process_sync(
        self,
        source: str,
        source_type: str,
        chunking_strategy: str = "basic",
        timeout: int = 30,
        **params
) -> Dict:
    """
    Synchronous process task that returns text directly (for real-time API)

    Args:
        source: Source file path, URL, or text content
        source_type: source of the file("local", "minio")
        chunking_strategy: Strategy for chunking the document
        timeout: Timeout for the operation
        **params: Additional parameters

    Returns:
        Dict containing the extracted text and metadata
    """
    start_time = time.time()
    task_id = self.request.id

    # Check if we're in a valid Celery context before updating state
    is_celery_context = hasattr(
        self, 'request') and self.request.id is not None

    # Update task state to PROCESSING only if in Celery context
    if is_celery_context:
        self.update_state(
            state=states.STARTED,
            meta={
                'source': source,
                'source_type': source_type,
                'task_name': 'process_sync',
                'start_time': start_time,
                'sync_mode': True
            }
        )

    logger.info(
        f"Synchronous processing file: {source} with strategy: {chunking_strategy}")

    # Get the data processor instance
    actor = get_ray_actor()

    try:
        # Process the file based on the source type
        if source_type == "local":
            # The unified actor call, mapping 'file' source_type to 'local' destination
            chunks_ref = actor.process_file.remote(
                source,
                chunking_strategy,
                destination=source_type,
                task_id=task_id,
                **params
            )