Metadata-Version: 2.4
Name: aria2-rust
Version: 0.3.9
Summary: Python SDK for aria2-rust JSON-RPC interface
Author: aria2-rust contributors
License: GPL-3.0-or-later
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25
Requires-Dist: typing-extensions>=4.0
Requires-Dist: websockets>=12.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.21; extra == 'dev'
Requires-Dist: pytest-cov>=4.0; extra == 'dev'
Requires-Dist: pytest>=7.0; extra == 'dev'
Requires-Dist: respx>=0.20; extra == 'dev'
Description-Content-Type: text/markdown

# aria2-rust

Python SDK for the aria2-rust JSON-RPC and WebSocket interface. The client
surface has complete type annotations and current binding coverage; full
original-client compatibility remains tracked in the repository matrix.

## Features

- ✅ **RPC Client Surface** - Task, status, option, session, and event methods
- ✅ **Dual Transport** - HTTP JSON-RPC and WebSocket support
- ✅ **Type Safe** - Full type annotations with mypy/pyright support
- ✅ **Error Handling** - Structured exception hierarchy
- ✅ **Auto Reconnection** - Exponential backoff (1s → 2s → 4s → 8s → 16s)
- ✅ **Event Subscription** - Real-time download event notifications
- ✅ **Async/Await** - Native asyncio support
- ✅ **Context Manager** - `async with` support for automatic cleanup
- ✅ **Authentication** - Token and Basic auth support
- ✅ **Production Ready** - Unit + Integration + E2E tests

## Installation

```bash
pip install aria2-rust
```

Or from source:

```bash
cd bindings/python
pip install -e ".[dev]"
```

## Quick Start

### Basic Usage

```python
import asyncio
from aria2_rust_client import Aria2Client

async def main():
    async with Aria2Client("http://localhost:6800/jsonrpc") as client:
        # Add a download task
        gid = await client.add_uri(["http://example.com/file.zip"])
        print(f"Download started: {gid}")
        
        # Check status
        status = await client.tell_status(gid)
        print(f"Status: {status.status}, Progress: {status.completed_length}/{status.total_length}")

asyncio.run(main())
```

### With Authentication

```python
# Token authentication
client = Aria2Client("http://localhost:6800/jsonrpc", token="my-secret-token")

# The token is automatically prepended to all RPC calls
gid = await client.add_uri(["http://example.com/file.zip"])
```

### WebSocket Event Subscription

```python
import asyncio
from aria2_rust_client import Aria2Client, EventType

async def main():
    async with Aria2Client("ws://localhost:6800/jsonrpc") as client:
        # Subscribe to all events
        async for event in client.subscribe_events():
            print(f"Event: {event.event_type}, GID: {event.gid}")
            
        # Or filter specific event types
        async for event in client.subscribe_events(filter=[EventType.DownloadStart, EventType.DownloadComplete]):
            print(f"Download event: {event.event_type}")

asyncio.run(main())
```

## API Reference

### Aria2Client

#### Constructor

```python
Aria2Client(
    url: str = "http://localhost:6800/jsonrpc",
    token: Optional[str] = None,
    timeout: float = 30.0
)
```

**Parameters:**
- `url` - RPC endpoint URL (http:// or ws://)
- `token` - Authentication token (optional)
- `timeout` - Request timeout in seconds

#### Methods

The async client exposes the standard aria2 RPC methods listed below. Methods
not listed here can still be called through the transport layer.

**Task Management:**
- `add_uri(uris, options=None, position=None)` - Add HTTP/FTP download
- `add_torrent(torrent, options=None, web_seed_uris=None, position=None)` - Add BitTorrent download
- `add_metalink(metalink, options=None, position=None)` - Add Metalink download; returns all created GIDs
- `remove(gid)` - Remove download
- `pause(gid)` - Pause download
- `unpause(gid)` - Resume download
- `force_pause(gid)` - Force pause
- `force_remove(gid)` - Force remove
- `pause_all()` / `force_pause_all()` / `unpause_all()` - Batch task control
- `change_position(gid, position, mode)` - Change queue position
- `change_uri(gid, file_index, delete_uris, add_uris, position=None)` - Replace task URIs

**Status Queries:**
- `tell_status(gid, keys=None)` - Get task status
- `get_files(gid)` - Get file paths, sizes, completion, and URI metadata for a task
- `get_uris(gid)` / `get_servers(gid)` / `get_peers(gid)` - Get transfer connection metadata
- `tell_active(keys=None)` - Get active tasks
- `tell_waiting(offset, num, keys=None)` - Get waiting tasks
- `tell_stopped(offset, num, keys=None)` - Get stopped tasks
- `get_global_stat()` - Get global statistics

**Session and browser context:**
- `update_browser_context(context)` / `clear_browser_context()`

**System methods:**
- `system_multicall(calls)` / `system_list_methods()` / `system_list_notifications()`

`get_files(gid)` is the direct binding for aria2's `aria2.getFiles` method:

```python
files = await client.get_files(gid)
for file in files:
    print(file.path, file.length)
```

For HTTP/FTP URLs, the length may be unknown until the metadata probe has
completed. For magnet links, the file list is unavailable until metadata
exchange completes. A local torrent is parsed when it is added, so its file
metadata can be queried even when the task is created with `pause=true`.

**Options:**
- `get_global_option()` - Get global options
- `change_global_option(options)` - Change global options
- `get_option(gid)` - Get task options
- `change_option(gid, options)` - Change task options

**System:**
- `get_version()` - Get version info
- `get_session_info()` - Get session info
- `shutdown()` - Graceful shutdown
- `force_shutdown()` - Force shutdown
- `save_session()` - Save session
- `purge_download_result()` - Purge completed results
- `remove_download_result(gid)` - Remove specific result

**Event Subscription:**
- `subscribe_events(filter=None)` - Subscribe to download events

**Lifecycle:**
- `close()` - Close connection
- `async with` - Context manager support

### Types

#### StatusInfo

```python
@dataclass
class StatusInfo:
    gid: str
    total_length: Optional[int]
    completed_length: Optional[int]
    download_speed: Optional[int]
    upload_speed: Optional[int]
    error_code: Optional[str]
    status: DownloadStatus
    dir: Optional[str]
    files: List[FileInfo]
```

#### GlobalStat

```python
@dataclass
class GlobalStat:
    download_speed: int
    upload_speed: int
    num_active: int
    num_waiting: int
    num_stopped: int
```

#### DownloadEvent

```python
@dataclass
class DownloadEvent:
    event_type: EventType
    gid: str
    error_code: Optional[int]
    files: List
```

#### Enums

```python
class EventType:
    DOWNLOAD_START = "aria2.onDownloadStart"
    DOWNLOAD_PAUSE = "aria2.onDownloadPause"
    DOWNLOAD_STOP = "aria2.onDownloadStop"
    DOWNLOAD_COMPLETE = "aria2.onDownloadComplete"
    DOWNLOAD_ERROR = "aria2.onDownloadError"
    BT_DOWNLOAD_COMPLETE = "aria2.onBtDownloadComplete"
    BT_DOWNLOAD_ERROR = "aria2.onBtDownloadError"

class DownloadStatus:
    ACTIVE = "active"
    WAITING = "waiting"
    PAUSED = "paused"
    ERROR = "error"
    COMPLETE = "complete"
    REMOVED = "removed"
```

### Errors

```python
class Aria2Error(Exception):
    code: int
    message: str

class ConnectionError(Aria2Error):  # code = -2
class AuthError(Aria2Error):  # code = -3
class RpcError(Aria2Error):  # code from JSON-RPC response
class TimeoutError(Aria2Error):  # code = -4
```

## Examples

### Download Progress Monitoring

```python
import asyncio
from aria2_rust_client import Aria2Client

async def download_with_progress(url, output_path=None):
    async with Aria2Client() as client:
        gid = await client.add_uri([url])
        
        while True:
            status = await client.tell_status(gid)
            progress = int(status.completed_length) / int(status.total_length) * 100
            speed = int(status.download_speed) / 1024  # KB/s
            
            print(f"Progress: {progress:.1f}%, Speed: {speed:.1f} KB/s")
            
            if status.status in ['complete', 'error', 'removed']:
                break
                
            await asyncio.sleep(1)
        
        return status

asyncio.run(download_with_progress("http://example.com/largefile.zip"))
```

### Batch Download

```python
import asyncio
from aria2_rust_client import Aria2Client

async def batch_download(urls, max_concurrent=5):
    async with Aria2Client() as client:
        # Add multiple tasks
        tasks = [client.add_uri([url]) for url in urls]
        gids = await asyncio.gather(*tasks)
        
        print(f"Added {len(gids)} tasks")
        
        # Monitor all tasks
        while True:
            statuses = await asyncio.gather(*[client.tell_status(gid) for gid in gids])
            
            active = sum(1 for s in statuses if s.status == 'active')
            complete = sum(1 for s in statuses if s.status == 'complete')
            error = sum(1 for s in statuses if s.status == 'error')
            
            print(f"Active: {active}, Complete: {complete}, Error: {error}")
            
            if complete + error == len(gids):
                break
                
            await asyncio.sleep(2)

urls = [
    "http://example.com/file1.zip",
    "http://example.com/file2.zip",
    "http://example.com/file3.zip",
]

asyncio.run(batch_download(urls))
```

### Torrent Download

```python
import asyncio
from aria2_rust_client import Aria2Client

async def download_torrent(torrent_path):
    async with Aria2Client() as client:
        with open(torrent_path, 'rb') as f:
            torrent_data = f.read()
        
        gid = await client.add_torrent(torrent_data)
        print(f"Torrent added: {gid}")
        
        # Monitor progress
        while True:
            status = await client.tell_status(gid)
            if status.status in ['complete', 'error']:
                break
            await asyncio.sleep(5)

asyncio.run(download_torrent("example.torrent"))
```

## Testing

### Run All Tests

```bash
# Unit tests
pytest tests/unit/ -v

# Integration tests (requires mock server)
pytest tests/integration/ -v

# E2E tests (requires aria2-rust binary)
pytest tests/e2e/ -v

# All tests
pytest tests/ -v
```

### Test Coverage

```bash
pytest tests/ --cov=aria2_rust_client --cov-report=html
```

## Development

### Setup Development Environment

```bash
pip install -e ".[dev]"
```

### Run Type Checker

```bash
mypy aria2_rust_client
```

### Run Linter

```bash
flake8 aria2_rust_client
```

## Requirements

- Python 3.9+
- httpx >= 0.25
- websockets >= 12.0

### Development Dependencies

- pytest >= 7.0
- pytest-asyncio >= 0.21
- respx >= 0.20
- mypy (optional)
- flake8 (optional)

## License

GPL-2.0-or-later

## Contributing

1. Fork the repository
2. Create a feature branch
3. Run tests: `pytest tests/ -v`
4. Submit a pull request

## Support

For issues and feature requests, please open an issue on the GitHub repository.
