feat: Data Source Connectors (Beyond CSV)

[aGallea]

Problem

Input is limited to local CSV files. Real-world data lives in databases, APIs, S3 buckets, and other formats. The current architecture tightly couples the indexing pipeline to CSV parsing, making it difficult to ingest data from other sources without significant refactoring.

Proposed Solution

Add pluggable data source connectors with a common interface. The current CSV reader becomes one implementation of a DataSource protocol. Additional connectors for JSON/JSONL, databases, and remote URLs extend the tool to real-world data pipelines.

Proposed Connectors (Priority Order)

1. JSON / JSONL — Flat JSON arrays or newline-delimited JSON files
2. URL / API — Fetch data from a REST endpoint that returns JSON
3. Database — PostgreSQL / MySQL via connection string with SQL query
4. S3 / GCS — Cloud storage bucket paths (stretch goal)

Implementation Details

Backend

New file: embedding_cluster/data_sources/__init__.py

• Export the protocol and factory function

New file: embedding_cluster/data_sources/base.py

• Define DataSource protocol:

  from collections.abc import AsyncIterator
  from typing import Protocol, Any
  
  class DataSource(Protocol):
      async def iter_rows(self) -> AsyncIterator[dict[str, Any]]:
          """Yield rows as dictionaries."""
          ...
  
      async def count(self) -> int | None:
          """Return total row count if known, None otherwise."""
          ...
  
      def validate(self) -> None:
          """Validate configuration. Raise ValueError if invalid."""
          ...

New file: embedding_cluster/data_sources/csv_source.py

• CsvDataSource — Extract current CSV parsing from indexer.py into this class
• Implements DataSource protocol
• Supports start_line, end_line parameters (existing functionality)

New file: embedding_cluster/data_sources/json_source.py

• JsonDataSource — Parse JSON array or JSONL files
• Auto-detect format (array vs newline-delimited)
• Support nested field access via dot notation (e.g., "metadata.title")

New file: embedding_cluster/data_sources/url_source.py

• UrlDataSource — Fetch JSON from a REST endpoint
• Support pagination (offset/limit, cursor, link-header)
• Configurable headers (for API keys)
• Use aiohttp (existing dependency) for async fetching

New file: embedding_cluster/data_sources/database_source.py

• DatabaseDataSource — Connect to PostgreSQL/MySQL
• Takes connection string + SQL query
• Stream results to avoid loading entire dataset into memory
• Use asyncpg for PostgreSQL (new optional dependency)

New file: embedding_cluster/data_sources/factory.py

• Factory function to create the right data source:

  def create_data_source(source_type: str, config: dict[str, Any]) -> DataSource:
      if source_type == "csv":
          return CsvDataSource(**config)
      elif source_type == "json":
          return JsonDataSource(**config)
      # ...

Modify: embedding_cluster/settings.py

• Add data_source_type: str = Field(default="csv", description="Data source type: csv, json, url, database")
• Add data_source_config: dict[str, Any] | None = Field(default=None, description="Data source specific configuration")
• Keep LOCAL_CSV_FILENAME as backward-compatible alias for csv source

Modify: embedding_cluster/indexer.py

• Refactor to accept a DataSource instead of reading CSV directly
• Replace direct CSV parsing with data_source.iter_rows()
• Keep backward compatibility: if LOCAL_CSV_FILENAME is set and no data_source_type, default to CSV

Modify: embedding_cluster/server/routes/index.py

• Accept data_source_type and data_source_config in index request
• Create appropriate data source via factory
• For file-based sources (CSV, JSON), handle upload flow

Modify: embedding_cluster/server/models.py

• Add data_source_type and data_source_config to IndexRequest
• Add per-source config models for validation

Frontend

Modify: frontend/src/pages/IndexPage.tsx

• Add data source type selector at the top of the form
• Dynamic config fields per source type:
  • CSV: File upload (current behavior)
  • JSON: File upload + format toggle (array vs JSONL)
  • URL: URL input, headers key-value editor, pagination config
  • Database: Connection string input, SQL query textarea, "Test Connection" button
• Preview adapts to show rows from any source type

New file: frontend/src/components/index/DataSourceConfig.tsx

• Renders source-specific configuration form
• Validates required fields per source type

Modify: frontend/src/api/index.ts

• Update index request to include data source configuration

Modify: frontend/src/types/index.ts

• Add data source types and config interfaces

Dependency Management

• JSON/JSONL: No new dependencies (stdlib json)
• URL: aiohttp already in dependencies
• Database: asyncpg as optional dependency
• S3/GCS: aiobotocore / gcloud-aio-storage as optional dependencies

Add optional dependency groups in pyproject.toml:

[project.optional-dependencies]
database = ["asyncpg>=0.29"]
cloud = ["aiobotocore>=2.0"]

Testing Requirements

Full-coverage tests are required for all new code.

Backend Tests

New file: tests/test_data_sources/test_csv_source.py

• Test CSV parsing matches current behavior exactly
• Test start_line / end_line
• Test missing file error
• Test malformed CSV handling
• Test empty CSV

New file: tests/test_data_sources/test_json_source.py

• Test JSON array parsing
• Test JSONL parsing
• Test auto-detection of format
• Test nested field access
• Test malformed JSON error
• Test empty file

New file: tests/test_data_sources/test_url_source.py

• Test fetching from URL (mock aiohttp)
• Test pagination (offset/limit)
• Test custom headers
• Test HTTP error handling (404, 500, timeout)
• Test empty response

New file: tests/test_data_sources/test_database_source.py

• Test database query execution (mock asyncpg)
• Test streaming results
• Test connection error handling
• Test invalid SQL handling
• Test empty result set

New file: tests/test_data_sources/test_factory.py

• Test factory creates correct data source for each type
• Test unknown source type raises ValueError
• Test config validation per source type

Modify: tests/test_indexer.py

• Test indexer works with DataSource protocol (not just CSV)
• Test backward compatibility (LOCAL_CSV_FILENAME still works)

Frontend Tests

• DataSourceConfig: renders correct fields per source type
• Validation: required fields enforced
• IndexPage: source type switching works
• Preview: adapts to non-CSV sources

Acceptance Criteria

• CSV source maintains full backward compatibility
• JSON/JSONL files can be used as data source
• REST API endpoints can be fetched as data source
• Database queries can be used as data source (PostgreSQL)
• Data source type is configurable via env var (CLI) and UI (Server)
• Preview works for all source types
• New dependencies are optional (not required for CSV-only usage)
• All new backend code has full test coverage
• All new frontend components have tests

Notes

• After implementation, update README.md to document all data source types (add new section, update parameter tables)
• Backward compatibility is critical — existing CSV workflows must not break
• Database connector should never store credentials in plain text — use connection strings or env vars
• Consider adding a DataSource.schema() method that returns field names/types for dynamic form generation
• S3/GCS is a stretch goal — implement CSV, JSON, URL, Database first