Design CSV upload endpoint with GPT classification

Problem Statement

You are building a backend service for an AI data-labeling platform. Clients upload CSV files containing unlabeled text snippets, which must be validated, classified via an external GPT-based classification client, and saved locally as structured JSON for downstream evaluation.

Implement the CSVClassificationPipeline class with the following specifications:

1. Parse & Validate CSV: The input is raw bytes representing a CSV file. The CSV must have headers id (string) and text (string). Skip rows where id or text is missing or whitespace-only, incrementing a skipped counter.
2. Classify: For each valid row, call classifier.classify(text). The classifier returns a dictionary {"label": str, "confidence": float}. If the classifier throws an exception, capture the error for that row ("error": str) without failing the entire file.
3. Persist: Write the final processed output to a local JSON file at output_path. The JSON output should contain:
   • summary: total rows, successful classifications, failed classifications, skipped rows.
   • results: a list of records containing id, text, and either label + confidence or error.
4. Return Value: Return the summary dictionary.

Interface

from typing import Any, Dict, Protocol

class LLMClassifier(Protocol):
    def classify(self, text: str) -> Dict[str, Any]:
        """Returns: {'label': str, 'confidence': float}"""
        ...

class CSVClassificationPipeline:
    def __init__(self, classifier: LLMClassifier):
        self.classifier = classifier

    def process_file(self, file_content: bytes, output_path: str) -> Dict[str, Any]:
        """
        Processes the raw CSV bytes, persists results to output_path, and returns summary.
        """
        pass

Constraints

• File content size is up to 10 MB (approximately 5,000 rows).
• Text snippets may contain UTF-8 characters, commas, and embedded newlines.
• If the CSV header is missing required columns (id, text) or the file cannot be parsed, raise a ValueError.

Example

csv_data = b"""id,text
1,"Great product! Highly recommend."
2,
3,"Broken on arrival."
"""

# classifier returns:
# classify("Great product! Highly recommend.") -> {"label": "positive", "confidence": 0.98}
# classify("Broken on arrival.") -> raises RuntimeError("API timeout")

pipeline = CSVClassificationPipeline(classifier)
summary = pipeline.process_file(csv_data, "results/batch_1.json")

Output Summary:

{
    "total_rows": 3,
    "successful": 1,
    "failed": 1,
    "skipped": 1
}

Persisted JSON (results/batch_1.json):

{
  "summary": {"total_rows": 3, "successful": 1, "failed": 1, "skipped": 1},
  "results": [
    {"id": "1", "text": "Great product! Highly recommend.", "label": "positive", "confidence": 0.98},
    {"id": "3", "text": "Broken on arrival.", "error": "API timeout"}
  ]
}