Metadata-Version: 2.4
Name: temporaliox
Version: 0.3.0
Summary: A helper library for Temporal.io that enables separation of activity declarations from implementations
Author: Igor Kozyrenko
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Topic :: System :: Distributed Computing
License-File: LICENSE
Requires-Dist: temporalio>=1.5.0
Requires-Dist: pytest>=8.0.0 ; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.0 ; extra == "dev"
Requires-Dist: ruff>=0.1.0 ; extra == "dev"
Requires-Dist: black>=23.0.0 ; extra == "dev"
Requires-Dist: build>=0.10.0 ; extra == "dev"
Project-URL: Homepage, https://github.com/ikseek/temporaliox
Project-URL: Repository, https://github.com/ikseek/temporaliox
Provides-Extra: dev

# TemporalioX

A helper library for Temporal.io that enables separation of activity declarations from their implementations.

## Installation

```bash
pip install temporaliox
```

## Usage

### 1. Declare Activities

In your `activities.py` file, declare activities with their Temporal configuration:

```python
from temporaliox.activity import decl
from datetime import timedelta

@decl(task_queue="playwright", start_to_close_timeout=timedelta(seconds=60))
def generate_report(user_name: str, action: str) -> str:
    pass
```

### 2. Define Activity Implementations

In your `activities_def.py` file, provide the actual implementations:

```python
from .activities import generate_report

@generate_report.defn
def generate_report_def(user_name: str, action: str) -> str:
    return f"{action}, {user_name}!"
```

### 3. Use Activities in Workflows

In your `workflows.py` file, call activities as normal async functions:

```python
from temporalio import workflow
from .activities import generate_report

@workflow.defn()
class GenerateReport:
    @workflow.run
    async def run(self):
        # This will call temporalio.workflow.execute_activity under the hood
        report = await generate_report("Alice", "Hello")
        # report will be "Hello, Alice!"
        return report
```

### Starting Activities

You can also start activities asynchronously using the `.start()` method:

```python
# Start activity and get handle
handle = generate_report.start("Bob", "Goodbye")
# Wait for result later
result = await handle
```

### 4. Set Up Workers

Use `activities_for_queue()` to automatically collect all implemented activities for a specific task queue:

```python
from temporalio.worker import Worker
from temporaliox.activity import activities_for_queue

async def main():
    worker = Worker(
        client,
        task_queue="playwright",
        workflows=[GenerateReport],
        activities=activities_for_queue("playwright"),  # Automatically gets all activities for this queue
    )
    await worker.run()
```

The `activities_for_queue()` function will:
- Return all implemented activities for the specified queue
- Raise a `ValueError` if any declared activities are missing implementations
- Return an empty list if no activities exist for the queue

## Features

- **Separation of Concerns**: Declare activity signatures and Temporal configuration separately from implementations
- **Type Safety**: Full type hints support for better IDE experience
- **Automatic Registration**: Activities are automatically registered with Temporal using the `@ActivityDeclaration.defn` decorator
- **Flexible Execution**: Support for both `await` (execute_activity) and `.start()` (start_activity) patterns
- **Queue-based Organization**: Activities are organized by task queue for easy worker setup

## How It Works

1. `@decl()` creates an `ActivityDeclaration` that holds the activity configuration and registers it as undefined
2. `@ActivityDeclaration.defn` decorates the implementation, registers it with Temporal, and moves it to the implemented registry
3. When called in a workflow, the activity uses `workflow.execute_activity` or `workflow.start_activity` with the configured options
4. `activities_for_queue()` collects all implemented activities for a queue, ensuring no undefined activities remain

## Development

### Running Tests

```bash
# Install development dependencies
pip install -e ".[dev]"

# Start Temporal server (required for integration tests)
temporal server start-dev

# Run tests
pytest
```
