Skip to content

Batch Service Example

A complete example of a production-ready IVCAP batch service.

Overview

This service processes batches of text items with validation, transformation, and error handling.

Code

from pydantic import BaseModel, Field, field_validator
from ivcap_service import (
    Service, ServiceContact, ServiceLicense,
    JobContext, start_batch_service,
    getLogger, logging_init, with_schema,
)
import time

# Initialize logging
logging_init()
logger = getLogger("batch_service")

# Define service
service = Service(
    name="Batch Text Processor",
    contact=ServiceContact(name="Developer", email="dev@example.com"),
    license=ServiceLicense(name="MIT", url="https://opensource.org/license/MIT"),
)

# Helper item type (no schema needed — embedded in BatchRequest)
class Item(BaseModel):
    id: str
    text: str = Field(min_length=1, max_length=1000)

# Request schema
@with_schema("urn:sd:schema:batch-text-processor.request.1")
class BatchRequest(BaseModel):
    items: list[Item] = Field(min_items=1)
    uppercase: bool = Field(default=False)

# Helper result item type
class ProcessedItem(BaseModel):
    id: str
    original: str
    processed: str
    word_count: int

# Result schema
@with_schema("urn:sd:schema:batch-text-processor.1")
class BatchResult(BaseModel):
    items: list[ProcessedItem]
    total_processed: int
    total_time_seconds: float

# Job processor
def process_job(req: BatchRequest, ctx: JobContext) -> BatchResult:
    """Process a batch of text items."""
    logger.info(f"Processing batch with {len(req.items)} items")
    start_time = time.time()

    results = []

    # Step 1: Validate inputs
    with ctx.report.step("validate", message="Validating inputs") as step:
        for item in req.items:
            if not item.text.strip():
                raise ValueError(f"Item {item.id}: empty text")
        step.finished(message=f"Validated {len(req.items)} items")

    # Step 2: Process items
    with ctx.report.step("process", message="Processing items") as step:
        for i, item in enumerate(req.items):
            # Transform
            processed_text = item.text.upper() if req.uppercase else item.text
            word_count = len(processed_text.split())

            results.append(ProcessedItem(
                id=item.id,
                original=item.text,
                processed=processed_text,
                word_count=word_count
            ))

            # Report progress
            if (i + 1) % 10 == 0:
                step.info(event={
                    "processed": i + 1,
                    "total": len(req.items),
                    "percentage": ((i + 1) / len(req.items)) * 100
                })

        step.finished(message=f"Processed {len(results)} items")

    # Step 3: Summary
    duration = time.time() - start_time

    with ctx.report.step("summary") as step:
        step.info(event={
            "total_items": len(results),
            "duration_seconds": duration,
            "items_per_second": len(results) / duration
        })
        step.finished()

    return BatchResult(
        items=results,
        total_processed=len(results),
        total_time_seconds=duration
    )

if __name__ == "__main__":
    start_batch_service(service, process_job)

Testing

Create test_request.json:

{
    "items": [
        {"id": "1", "text": "Hello world"},
        {"id": "2", "text": "IVCAP service"},
        {"id": "3", "text": "Batch processing"}
    ],
    "uppercase": true
}

Test locally:

python batch_service.py --test-file test_request.json

Expected output:

Processing batch with 3 items
Validated 3 items
Processed 3 items
...

Key Features

  1. Validation — Pydantic validates request automatically
  2. Progress Tracking — Reports progress with step events
  3. Error Handling — Explicit error reporting
  4. Metrics — Tracks processing time and statistics
  5. Logging — Structured logging for debugging

See Also