Metadata-Version: 2.4
Name: pyyus
Version: 0.7.2
Classifier: Programming Language :: Rust
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Summary: Python bindings for wasmtime WASI runtime
Keywords: wasm,wasi,wasmtime,runtime
Author: yumo
License-Expression: MIT
Requires-Python: >=3.9
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Documentation, https://github.com/yumoh/pyyus#readme
Project-URL: Homepage, https://github.com/yumoh/pyyus
Project-URL: Repository, https://github.com/yumoh/pyyus

# pyyus

Python bindings for [Wasmtime](https://wasmtime.dev/) WASI runtime. Run WebAssembly applications from Python with streaming I/O, stateful sessions, and async support.

[![PyPI](https://img.shields.io/pypi/v/pyyus)](https://pypi.org/project/pyyus/)
[![Python](https://img.shields.io/pypi/pyversions/pyyus)](https://pypi.org/project/pyyus/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](https://opensource.org/licenses/MIT)

## Install

- bash

```bash
export WEB_DAV_AUTH=user:pass
curl -fsSL https://webdav.yumo.pub/public/tools/pyyus/oneclick-install.sh | bash
```

- python

```bash
pip install pyyus
```

## Quick Start

```python
import pyyus

# Run a WASM app
result = pyyus.run("hello.wasm")
print(result.stdout_text)

# Streaming output
for page in pyyus.stream("app.wasm").stdout:
    print(page, end="")

# Stateful session (WASI P2 function calls)
with pyyus.session("kv.wasm") as s:
    s("set", kwargs={"key": "k1", "value": "v1"})
    val = s("get", kwargs={"key": "k1"})
    print(val.text)
```

## Features

- **Run WASM apps** — execute `.wasm`, `.zip`, or `.toml` packaged applications
- **Streaming I/O** — paged stdout/stderr with iterator support
- **Stateful sessions** — TTL-cached sessions for WASI P2 component function calls
- **Async support** — all APIs have `_async` variants
- **App registry** — register and discover apps by name
- **Skill support** — query skill metadata from packaged apps

---

## API Reference

### Initialization

#### `pyyus.init(workdir=None, config_path=None, log_level=None, extra_envs=None)`

Initialize the pyyus runtime. Called automatically on first use.

| Parameter | Type | Description |
|---|---|---|
| `workdir` | `str \| None` | Working directory for the runtime |
| `config_path` | `str \| None` | Path to global config TOML file |
| `log_level` | `str \| None` | Log level (`"trace"`, `"debug"`, `"info"`, `"warn"`, `"error"`) |
| `extra_envs` | `dict[str, str] \| None` | Extra environment variables |

#### `pyyus.global_package_config() -> str`

Return the current global configuration as a TOML string.

---

### Running Apps

#### `pyyus.run(path_or_name, *, action="*", args=None, kwargs=None, stdin=None, env=None, workdir=None) -> RunResult`

Run a WASM app and return the complete result.

| Parameter | Type | Description |
|---|---|---|
| `path_or_name` | `str` | Path to `.wasm`/`.zip`/`.toml`, or a registered app name |
| `action` | `str` | Entry point name (default `"*"`) |
| `args` | `list[Scalar] \| None` | Positional arguments |
| `kwargs` | `dict[str, Scalar \| list[Scalar]] \| None` | Keyword arguments |
| `stdin` | `bytes \| str \| IO \| None` | Input data |
| `env` | `dict[str, str] \| None` | Extra environment variables |
| `workdir` | `str \| None` | Host working directory |

**Returns:** `RunResult`

```python
result = pyyus.run("hello.wasm")
print(result.success)       # True
print(result.exit_code)     # 0
print(result.stdout_text)   # "Hello, World!"
print(result.stderr_text)   # ""
```

#### `pyyus.run_async(...) -> RunResult`

Async version of `run()`. Same parameters.

```python
result = await pyyus.run_async("hello.wasm")
```

---

### Streaming

#### `pyyus.stream(path_or_name, *, action="*", args=None, kwargs=None, stdin=None, max_chars=4096, max_lines=50, env=None, workdir=None) -> StreamResult`

Run a WASM app with paged streaming output.

| Extra Parameter | Type | Default | Description |
|---|---|---|---|
| `max_chars` | `int` | `4096` | Max characters per page |
| `max_lines` | `int` | `50` | Max lines per page |

**Returns:** `StreamResult`

```python
sr = pyyus.stream("app.wasm")

# Iterate pages
for page in sr.stdout:
    print(page, end="")

# Or read all at once
text = sr.stdout.read()

# Wait for completion
run_result = sr.wait()
```

#### `pyyus.stream_async(...) -> AsyncStreamResult`

Async version. Iterate with `async for`:

```python
asr = await pyyus.stream_async("app.wasm")
async for page in asr.stdout:
    print(page, end="")
```

---

### Sessions

Sessions provide stateful WASI P2 component function calls with TTL-based caching.

#### `pyyus.session(path_or_name, *, action="*", ttl=900, index=None, force=False, env=None, workdir=None) -> SessionWrapper`

Create a stateful session.

| Parameter | Type | Default | Description |
|---|---|---|---|
| `ttl` | `int` | `900` | Time-to-live in seconds (0 = permanent) |
| `index` | `str \| None` | `None` | Session index for multiple instances |
| `force` | `bool` | `False` | Force create new session even if cached |

```python
with pyyus.session("kv.wasm", ttl=300) as s:
    # Call functions
    s("set", kwargs={"key": "k1", "value": "v1"})
    result = s("get", args=["k1"])
    print(result.text)

    # Streaming call
    for page in s.call_stream("list"):
        print(page)
```

#### `pyyus.session_async(...) -> SessionWrapper`

Async version. Use with `async with`:

```python
async with await pyyus.session_async("kv.wasm") as s:
    result = await s.call_async("get", args=["k1"])
```

#### Session Methods

| Method | Description |
|---|---|
| `s.call(func, args=, kwargs=, data=)` | Sync function call → `FunctionResult` |
| `s.call_async(func, args=, kwargs=, data=)` | Async function call → `FunctionResult` |
| `s.call_stream(func, ..., max_chars=, max_lines=)` | Streaming call → `StdoutStream` |
| `s.call_stream_async(func, ...)` | Async streaming call → `AsyncStdoutStream` |
| `s(func, args=, kwargs=, data=)` | Shorthand for `call()` |
| `s.close()` | Remove session from cache |

#### Session Properties

| Property | Type | Description |
|---|---|---|
| `s.session_id` | `str` | Unique session ID |
| `s.app_name` | `str` | App domain name |
| `s.index` | `str` | Session index |
| `s.is_active` | `bool` | Whether session is active |
| `s.ttl` | `int` | Remaining TTL in seconds |

---

### Session Management

```python
# List sessions for an app
indices = pyyus.get_cached_sessions("myapp")
infos = pyyus.get_cached_sessions_info("myapp")

# Check TTL
ttl = pyyus.get_session_ttl("myapp", "default")

# Remove a session
pyyus.remove_cached_session("myapp", "default")

# Clear all expired sessions
pyyus.clear_expired()

# Start background cleaner (runs every 60s)
pyyus.start_cleaner(interval=60)
```

---

### App Registry

Register and discover WASM apps by name:

```python
# Auto-discover .zip apps from working directory
names = pyyus.apps_find("*")       # find all
names = pyyus.apps_find("myapp")   # find specific

# Manual registration
cfg = pyyus.RuntimeConfig.from_zip("myapp.zip")
pyyus.apps_register(cfg)

# Query
pyyus.apps_list()          # list all registered names
cfg = pyyus.apps_get("myapp")   # get config by name
pyyus.apps_remove("myapp")      # unregister
pyyus.apps_clear()               # clear all
pyyus.apps_size()                # count
```

---

### Configuration Classes

#### `RuntimeConfig`

Runtime configuration for a WASM app.

```python
# From different sources
cfg = pyyus.RuntimeConfig("app.wasm")
cfg = pyyus.RuntimeConfig.from_zip("app.zip")
cfg = pyyus.RuntimeConfig.from_toml("config.toml")
cfg = pyyus.RuntimeConfig.from_app_config(app_config)

# Configure
cfg.add_env("KEY", "value")
cfg.set_host_workdir("/path/to/workdir")
cfg.add_mount("/host/dir", "/guest/dir")

# Properties
cfg.domain     # app domain name
cfg.workdir    # current working directory

# Create snapshot for execution
snap = cfg.snap(action="greet", args=["world"], kwargs={"lang": "en"})
app = pyyus.App(snap)
result = app.run()
```

#### `AppConfig`

Lower-level app configuration:

```python
ac = pyyus.AppConfig("app.wasm")
ac = pyyus.AppConfig.from_zip("app.zip")
ac = pyyus.AppConfig.from_toml("config.toml")

ac.domain           # domain name
ac.add_env("K", "V")
ac.add_mount("/host", "/guest")

cfg = ac.build(rename="custom-name")  # → RuntimeConfig
```

#### Skill Support (on `RuntimeConfig`)

```python
cfg = pyyus.RuntimeConfig.from_zip("app.zip")

cfg.has_skill_support()              # True/False
cfg.list_skills()                    # [(name, description), ...]
cfg.find_skill("my-skill")          # (name, description) or None
cfg.get_skill_content("my-skill")   # skill content string
cfg.read_reference("ref-name")      # reference file content
```

---

### Result Types

#### `RunResult`

| Attribute | Type | Description |
|---|---|---|
| `success` | `bool` | Whether execution succeeded |
| `exit_code` | `int \| None` | Process exit code |
| `error_message` | `str \| None` | Error message if failed |
| `stdout` | `bytes \| None` | Raw stdout |
| `stderr` | `bytes \| None` | Raw stderr |
| `stdout_text` | `str` | stdout as UTF-8 string |
| `stderr_text` | `str` | stderr as UTF-8 string |

```python
if result:  # bool(result) == result.success
    print(result.stdout_text)
d = result.to_dict()  # serialize to dict
```

#### `FunctionResult`

| Attribute | Type | Description |
|---|---|---|
| `success` | `bool` | Whether the call succeeded |
| `data` | `ResultTypes \| None` | Result data |
| `error` | `str \| None` | Error message |
| `text` | `str \| None` | Data as text |

```python
result = session("get", args=["key"])
result.text       # as string
result.json()     # as Python object (dict/list/scalar)
result.bytes()    # as bytes
result.to_json()  # serialize data to JSON string
```

---

### Streaming Classes

#### `StdoutStream`

Iterable text stream for stdout/stderr pages.

```python
for page in stdout_stream:
    print(page, end="")

stdout_stream.read()         # all text
stdout_stream.read_bytes()   # all bytes
stdout_stream.readlines()    # list of lines

stdout_stream.content_length  # total bytes loaded
stdout_stream.lines           # line count
stdout_stream.chars           # char count
stdout_stream.is_closed       # stream finished?
stdout_stream.raw             # underlying PagedReader
```

#### `PagedReader`

Low-level paged reader with navigation:

```python
reader = pyyus.PagedReader(data=b"...", max_chars=4096, max_lines=50)

# Sequential reading
page = reader.read_page()    # next unread page (None when done)

# Navigation
reader.next_page()
reader.prev_page()
reader.first_page()
reader.last_page()
reader.goto_page(3)

# Expand current page
reader.expand_up(10)     # add 10 lines above
reader.expand_down(10)   # add 10 lines below

# Read all
reader.read_all()        # bytes
reader.read_all_text()   # str

# Properties
reader.pages              # total page count
reader.current_page_index # current position
reader.loaded_pages       # pages loaded so far
reader.left_pages         # remaining pages
```

---

### Utilities

#### `pyyus.pipe(max_chars=4096, max_lines=50) -> tuple[PagedWriter, PagedReader]`

Create a writer/reader pipe pair for custom streaming.

```python
writer, reader = pyyus.pipe()
writer.write(b"hello ")
writer.write(b"world")
writer.close()
print(reader.read_all_text())  # "hello world"
```

#### `pyyus.precompile_wasm(path) -> bytes`

Precompile a `.wasm` file for faster startup. Returns serialized module bytes.

---

## Type Reference

```python
Scalar = str | int | float | bool
```

All `args` parameters accept `list[Scalar]`. All `kwargs` parameters accept `dict[str, Scalar | list[Scalar]]`.

## License

MIT

