tosh 0.4.0
Declarative YAML framework for building PySpark custom data sources
Requires Python: >=3.12
Keywords: connectors, data-sources, declarative, etl, pyspark, yaml

# Tosh - Declarative YAML Framework for PySpark

> **Build PySpark custom data sources from YAML manifests** - Bringing Airbyte's low-code philosophy to the PySpark ecosystem.

[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/)
[![PySpark 4.0+](https://img.shields.io/badge/pyspark-4.0+-orange.svg)](https://spark.apache.org/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)

---

## Quick Start

### Installation

```bash
# Using uv (recommended)
uv add tosh

# Using pip
pip install tosh
```

### Hello World

Create a YAML manifest (`jsonplaceholder.yaml`):

```yaml
version: "1.0"

metadata:
  name: "jsonplaceholder"
  description: "JSONPlaceholder API connector"

streams:
  - name: "posts"
    
    retriever:
      requester:
        type: HttpRequester
        url_base: "https://jsonplaceholder.typicode.com"
        path: "/posts"
        http_method: "GET"
        
        authenticator:
          type: NoAuth
        
        paginator:
          type: NoPaginator
        
        record_selector:
          type: JsonPathSelector
          record_path: "$[*]"
    
    schema:
      type: object
      properties:
        id:
          type: integer
        userId:
          type: integer
        title:
          type: string
        body:
          type: string
```

Use it in PySpark:

```python
from pyspark.sql import SparkSession
from tosh import load_manifest

# Create Spark session
spark = SparkSession.builder.appName("tosh-demo").getOrCreate()

# Load and register the data source
datasource = load_manifest("jsonplaceholder.yaml")
spark.dataSource.register(datasource)

# Read data
df = spark.read.format("posts").load()
df.show()

# Output:
# +---+------+--------------------+--------------------+
# | id|userId|               title|                body|
# +---+------+--------------------+--------------------+
# |  1|     1|sunt aut facere r...|quia et suscipit ...|
# |  2|     1|qui est esse       |est rerum tempore...|
# ...
```

---

## Features

### Core Capabilities

- **Declarative YAML** - Define data sources in YAML instead of Python code
- **100+ Connector Patterns** - Authentication, pagination, partitioning out of the box
- **PySpark Native** - Full integration with PySpark's DataSource API
- **Batch & Streaming** - Support for both batch and streaming workloads
- **Parallel Processing** - Partition-aware for maximum parallelism
- **Template Support** - Jinja2 templates for dynamic configuration

### Built-in Components

| Component | Implementations |
|-----------|----------------|
| **Authentication** | NoAuth, API Key, Bearer, Basic, OAuth |
| **Pagination** | Page increment, Offset, Cursor-based |
| **Partitioning** | List, Substream, Date range |
| **Record Selection** | JSONPath, XPath, CSV |
| **Error Handling** | Retry, Backoff, Custom handlers |

---

## Examples

### Example 1: GitHub API with Authentication & Pagination

```yaml
version: "1.0"

metadata:
  name: "github"
  description: "GitHub API connector"

streams:
  - name: "pull_requests"
    primary_key: "id"
    
    retriever:
      requester:
        type: HttpRequester
        url_base: "https://api.github.com"
        path: "/repos/{{ config.repository }}/pulls"
        http_method: "GET"
        
        headers:
          Accept: "application/vnd.github+json"
        
        request_params:
          state: "all"
          per_page: "100"
        
        # Bearer token authentication
        authenticator:
          type: BearerAuthenticator
          token: "{{ config.api_token }}"
        
        # Page-based pagination
        paginator:
          type: PageIncrement
          page_size: 100
          page_param: "page"
          start_page: 1
        
        record_selector:
          type: JsonPathSelector
          record_path: "$[*]"
    
    schema:
      type: object
      properties:
        id:
          type: integer
        title:
          type: string
        state:
          type: string
        created_at:
          type: string

spec:
  type: object
  required:
    - api_token
    - repository
  properties:
    api_token:
      type: string
      title: "GitHub API Token"
    repository:
      type: string
      title: "Repository (owner/repo)"
      examples:
        - "apache/spark"
```

**Usage:**

```python
datasource = load_manifest("github.yaml")
spark.dataSource.register(datasource)

df = spark.read.format("pull_requests") \
    .option("api_token", "ghp_xxx") \
    .option("repository", "apache/spark") \
    .load()

df.select("id", "title", "state").show()
```

### Example 2: API Key Authentication

```yaml
retriever:
  requester:
    authenticator:
      type: ApiKeyAuthenticator
      api_key: "{{ config.api_key }}"
      header: "X-API-Key"  # Inject into header
      # OR
      # param_name: "api_key"  # Inject into query param
```

### Example 3: Cursor-Based Pagination

```yaml
paginator:
  type: CursorPaginator
  cursor_param: "cursor"
  cursor_path: "$.pagination.next_cursor"
  inject_into: "request_parameter"
```

### Example 4: List Partitioning for Parallel Processing

```yaml
retriever:
  partition_router:
    type: ListPartitioner
    values: ["US", "EU", "APAC"]
    cursor_field: "region"
    param_name: "region"
    inject_into: "request_parameter"
```

This will create 3 partitions, fetching data for each region in parallel!

---

## Architecture

```
┌─────────────────────────────────────────┐
│         YAML Manifest                    │
│  (Declarative Configuration)             │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│      ManifestParser                      │
│  - Parse YAML                            │
│  - Resolve $ref                          │
│  - Interpolate templates                 │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│    DataSourceGenerator                   │
│  - Generate PySpark DataSource class     │
│  - Wire up components                    │
└──────────────┬──────────────────────────┘
               │
               ▼
┌─────────────────────────────────────────┐
│  Generated PySpark DataSource            │
│  ├─ HttpRequester                        │
│  │  ├─ Authenticator                     │
│  │  ├─ Paginator                         │
│  │  └─ RecordSelector                    │
│  └─ Partitioner                          │
└─────────────────────────────────────────┘
               │
               ▼
        PySpark DataFrame
```

---

## Development

### Setup

```bash
# Clone the repository
git clone https://github.com/yourusername/tosh.git
cd tosh

# Install with dev dependencies using uv
uv sync --dev

# Or using pip
pip install -e ".[dev]"
```

### Running Tests

```bash
# Run all tests with coverage
pytest

# Run specific test file
pytest tests/test_parser.py

# Run with verbose output
pytest -v

# Generate coverage report
pytest --cov=tosh --cov-report=html
```

### Code Quality

```bash
# Format code
ruff format .

# Lint code
ruff check .

# Type checking
mypy src/tosh
```

---

## Documentation

- **[User Guide](docs/USER_GUIDE.md)** - Complete guide for end users!
- **[Tutorial](docs/TUTORIAL.md)** - Step-by-step getting started guide.
- **[Examples Gallery](examples/README.md)** - GitHub, REST APIs, Substreams, and more.
- **[Best Practices](docs/BEST_PRACTICES.md)** - Security, performance, and reliability tips.
- [Architecture Overview](docs/architecture.md)
- [YAML Schema Reference](src/tosh/schemas/manifest_schema.yaml)

---

## Contributing

Contributions are welcome! Please see [CONTRIBUTING.md](CONTRIBUTING.md) for details.

---

## License

MIT License - see [LICENSE](LICENSE) for details.

---

## Acknowledgments

Inspired by [Airbyte](https://airbyte.com/)'s low-code connector framework and built for the [PySpark](https://spark.apache.org/) ecosystem.

---

## Star History

If you find Tosh useful, please consider giving it a star! ⭐

---

**Built with ❤️ for the data engineering community**