Metadata-Version: 2.4
Name: motia
Version: 1.0.0rc21
Summary: Motia framework for III Engine
Author: III
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Requires-Dist: iii-sdk==0.2.0
Requires-Dist: pydantic>=2.0
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: mypy>=1.8; extra == 'dev'
Requires-Dist: opentelemetry-api>=1.20; extra == 'dev'
Requires-Dist: opentelemetry-exporter-otlp>=1.20; extra == 'dev'
Requires-Dist: opentelemetry-sdk>=1.20; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.2; extra == 'dev'
Provides-Extra: otel
Requires-Dist: opentelemetry-api>=1.20; extra == 'otel'
Requires-Dist: opentelemetry-exporter-otlp>=1.20; extra == 'otel'
Requires-Dist: opentelemetry-sdk>=1.20; extra == 'otel'
Description-Content-Type: text/markdown

# Motia Framework for Python

High-level framework for building workflows with the III Engine.

## Installation

```bash
uv pip install motia
```

## Usage

### Defining a Step

```python
from motia import FlowContext, queue

config = {
    "name": "process-data",
    "triggers": [queue("data.created")],
    "enqueues": ["data.processed"],
}

async def handler(data: dict, ctx: FlowContext) -> None:
    ctx.logger.info("Processing data", data)
    await ctx.enqueue({"topic": "data.processed", "data": data})
```

### API Steps

```python
from motia import ApiRequest, ApiResponse, FlowContext, http

config = {
    "name": "create-item",
    "triggers": [http("POST", "/items")],
    "emits": ["item.created"],
}

async def handler(req: ApiRequest, ctx: FlowContext) -> ApiResponse:
    ctx.logger.info("Creating item", req.body)
    await ctx.enqueue({"topic": "item.created", "data": req.body})
    return ApiResponse(status=201, body={"id": "123"})
```

### Streams

```python
from motia import Stream

# Define a stream
todo_stream = Stream[dict]("todos")

# Use the stream
item = await todo_stream.get("group-1", "item-1")
await todo_stream.set("group-1", "item-1", {"title": "Buy milk"})
await todo_stream.delete("group-1", "item-1")
items = await todo_stream.get_group("group-1")
```


### Build & Publish
```bash
python -m build
uv publish --index cloudsmith dist/*
```


## Features

- Event-driven step definitions
- API route handlers
- Cron job support
- Stream-based state management
- Type-safe context with logging

## Testing

### Running Integration Tests

Integration tests require a running III Engine instance. Make sure to have it built or installed before running tests.

1. Install dev dependencies:
   ```bash
   cd motia && uv sync --all-extras

   ```

   ```
3. Run tests:
   ```bash
   uv run pytest
   ```

### Test Configuration

Tests use non-default ports to avoid conflicts:
- Engine WebSocket: `ws://localhost:49199`
- HTTP API: `http://localhost:3199`

Set `III_ENGINE_PATH` environment variable to point to the III engine binary.

### Test Coverage

The integration test suite covers:
- Bridge connection and function registration
- API triggers (HTTP endpoints)
- KV Server operations
- PubSub messaging
- Logging module
- Motia framework integration
- Stream operations (when available)
- State management (when available)
