Data Processor
API Documentation

Overview

The Data Processor API lets you upload CSV files and process them with pandas-backed operations. Upload a CSV to get its parsed structure, then run statistics, text processing, column management, or cleaning. Designed for data pipelines and lightweight analytics.

Base URL & Authentication

Base URL

https://odivora.com/api/v1

Authentication

Include your API credentials in every request header.

X-API-Key: your_api_key
X-API-Secret: your_secret
Content-Type: application/json

Use the credentials shown in your API Keys section. In production, store them in environment variables or a secure secret store rather than source code.

Get API Keys

Follow these steps to obtain your API credentials:

Endpoints

Health Check

Health check endpoint to verify the Data Processor service status, available backend, and supported formats.

Response

{
  "success": true,
  "status": "healthy",
  "data_processor_backends": {
    "pandas": true
  },
  "supported_formats": [".csv"],
  "max_file_size_mb": 50,
  "timestamp": "2026-08-07T12:00:00"
}

Code Examples

import requests

url = "https://odivora.com/api/v1/data_processor/health"

headers = {
    "X-API-Key": "your_api_key",
    "X-API-Secret": "your_secret"
}

res = requests.get(url, headers=headers)
print(res.json())

curl -X GET https://odivora.com/api/v1/data_processor/health \
-H "X-API-Key: your_api_key" \
-H "X-API-Secret: your_secret"

fetch('https://odivora.com/api/v1/data_processor/health', {
  method: 'GET',
  headers: {
    'X-API-Key': 'your_api_key',
    'X-API-Secret': 'your_secret'
  }
})
.then(res => res.json())
.then(console.log);

Upload CSV

Upload a CSV file and receive its parsed structure (columns, rows, shape). Upload the file as multipart/form-data.

Form Parameters

Response

{
  "success": true,
  "data": {
    "columns": ["name", "age", "city"],
    "data": [
      ["name", "age", "city"],
      ["Alice", 30, "New York"],
      ["Bob", 25, "London"]
    ],
    "shape": [2, 3],
    "info": "2 rows × 3 columns"
  }
}

cURL Example

curl -X POST "https://odivora.com/api/v1/data_processor/upload" \
  -H "X-API-Key: your_api_key" \
  -H "X-API-Secret: your_secret" \
  -F "file=@data.csv"

Code Examples

import requests

url = "https://odivora.com/api/v1/data_processor/upload"

headers = {
    "X-API-Key": "your_api_key",
    "X-API-Secret": "your_secret"
}

with open("data.csv", "rb") as f:
    files = {"file": f}
    response = requests.post(url, headers=headers, files=files)

if response.status_code == 200:
    result = response.json()
    print("Columns:", result.get("data", {}).get("columns"))
    print("Info:", result.get("data", {}).get("info"))
else:
    print(response.json())

const form = new FormData();
form.append('file', fileInput.files[0]);

fetch('https://odivora.com/api/v1/data_processor/upload', {
  method: 'POST',
  headers: {
    'X-API-Key': 'your_api_key',
    'X-API-Secret': 'your_secret'
  },
  body: form
})
.then(res => res.json())
.then(data => console.log(data.data.info));

Process Data

Run a data operation on CSV data. Send the request as JSON with Content-Type: application/json. Use the data array returned by the upload endpoint.

JSON Parameters

Cleaning example

Cleaning can remove duplicate rows and fill missing values. The response includes original_rows, new_rows, operations_applied, and the processed data.

{
  "data": [["Name", "Age", "City"], ["Alice", 30, "Nairobi"], ["Alice", 30, "Nairobi"], ["Bob", null, "Nairobi"]],
  "operation": "cleaning",
  "remove_duplicates": true,
  "fill_rules": {
    "Age": {"method": "mean"},
    "City": {"method": "mode"}
  }
}

cURL Example

curl -X POST "https://odivora.com/api/v1/data_processor/process" \
  -H "X-API-Key: your_api_key" \
  -H "X-API-Secret: your_secret" \
  -H "Content-Type: application/json" \
  -d '{
    "data": [
      ["name", "age", "city"],
      ["Alice", 30, "New York"],
      ["Bob", 25, "London"]
    ],
    "operation": "statistics",
    "selected_columns": ["age"],
    "selected_metrics": ["mean", "count"]
  }'

Code Examples

import requests

url = "https://odivora.com/api/v1/data_processor/process"

headers = {
    "X-API-Key": "your_api_key",
    "X-API-Secret": "your_secret",
    "Content-Type": "application/json"
}

payload = {
    "data": [
        ["name", "age", "city"],
        ["Alice", 30, "New York"],
        ["Bob", 25, "London"]
    ],
    "operation": "statistics",
    "selected_columns": ["age"],
    "selected_metrics": ["mean", "median", "count"]
}

response = requests.post(url, headers=headers, json=payload)

if response.status_code == 200:
    print(response.json())
else:
    print(response.json())

const payload = {
  data: [
    ['name', 'age', 'city'],
    ['Alice', 30, 'New York'],
    ['Bob', 25, 'London']
  ],
  operation: 'statistics',
  selected_columns: ['age'],
  selected_metrics: ['mean', 'count']
};

fetch('https://odivora.com/api/v1/data_processor/process', {
  method: 'POST',
  headers: {
    'X-API-Key': 'your_api_key',
    'X-API-Secret': 'your_secret',
    'Content-Type': 'application/json'
  },
  body: JSON.stringify(payload)
})
.then(res => res.json())
.then(data => console.log(data));

Python SDK

Install the only dependency with pip install requests, then save the example as data_processor_example.py and run python data_processor_example.py. Replace the two placeholder credentials before running it.

import csv
import os
import requests

BASE_URL = "https://odivora.com/api/v1"
API_KEY = "YOUR_API_KEY"
API_SECRET = "YOUR_API_SECRET"
HEADERS = {"X-API-Key": API_KEY, "X-API-Secret": API_SECRET}
SAMPLE_ROWS = [
    ["Name", "Age", "City", "Salary"],
    ["Alice", 30, "Nairobi", 50000],
    ["Bob", 25, "Nairobi", 45000],
    ["Alice", 30, "Nairobi", 50000],
    ["Carol", "", "Mombasa", 60000],
]


def show_error(response):
    try:
        detail = response.json()
    except ValueError:
        detail = response.text
    labels = {401: "Authentication failure", 400: "Bad request",
              413: "File too large", 429: "Rate limit exceeded",
              500: "Server error"}
    print(f"{labels.get(response.status_code, 'Request failed')} "
          f"({response.status_code}): {detail}")


def request(method, path, **kwargs):
    response = requests.request(method, f"{BASE_URL}{path}", headers=HEADERS,
                                timeout=120, **kwargs)
    if not response.ok:
        show_error(response)
        return None
    return response.json()


def ensure_sample_csv(csv_path):
    directory = os.path.dirname(csv_path)
    if directory:
        os.makedirs(directory, exist_ok=True)
    with open(csv_path, "w", encoding="utf-8", newline="") as output:
        csv.writer(output).writerows(SAMPLE_ROWS)


def upload_csv(csv_path):
    with open(csv_path, "rb") as source:
        return request("POST", "/data_processor/upload", files={"file": source})


def process_data(data, operation, **params):
    payload = {"data": data, "operation": operation, **params}
    return request("POST", "/data_processor/process", json=payload)


if __name__ == "__main__":
    print("=== ODIVORA DATA PROCESSOR API TEST ===")
    ensure_sample_csv("sample_data.csv")

    health = request("GET", "/data_processor/health")
    if health:
        print("[1] Health Check       ✓", health["status"], health["data_processor_backends"])

    uploaded = upload_csv("sample_data.csv")
    if uploaded:
        data = uploaded["data"]["data"]
        print("[2] CSV Upload         ✓", uploaded["data"]["columns"], uploaded["data"]["shape"])

        statistics = process_data(data, "statistics", selected_columns=["Age", "Salary"],
                                  selected_metrics=["mean", "median", "count"])
        if statistics:
            print("[3] Statistics         ✓", statistics["data"]["results"])

        added = process_data(data, "add_column", column_name="Bonus")
        if added:
            print("[4] Add Column         ✓", added["data"]["data"][0])

        removed = process_data(data, "remove_columns", selected_columns=["Salary"])
        if removed:
            print("[5] Remove Columns     ✓", removed["data"]["data"][0])

        text = process_data(data, "text_processing", selected_columns=["Name", "City"],
                            text_operation="uppercase")
        if text:
            print("[6] Text Processing    ✓", text["data"]["data"][1:3])

        cleaning = process_data(data, "cleaning", remove_duplicates=True,
                                fill_rules={"Age": {"method": "mean"},
                                            "City": {"method": "mode"}})
        if cleaning:
            result = cleaning["data"]
            print("[7] Cleaning           ✓", result["new_rows"], result["operations_applied"])

    print("All tests completed.")

# Keep credentials out of source control in production: use environment
# variables or another secure secret store.

Errors & Status Codes

Error Response Format

{
  "success": false,
  "error": "Only CSV files supported"
}

Status Codes

Limitations