osiiso

Structured task queues for Python — one API across asyncio, threads, and processes.

---

What is osiiso?

osiiso gives you one compact queue API for three execution backends:

Backend	Class	Best for
Asyncio	AsyncQueue	HTTP clients, async databases, websockets, API fan-out
Threads	ThreadQueue	Blocking I/O, synchronous SDKs, filesystem work, SQLite
Processes	ProcessQueue	CPU-heavy computation, parsing, scoring, analytics

All three queues share the same shape: submit tasks → configure options → run → inspect results.

import asyncio
import osiiso


async def fetch(name: str) -> str:
    await asyncio.sleep(0.1)
    return f"fetched {name}"


async def main():
    async with osiiso.AsyncQueue(workers=4) as q:
        q.submit(fetch, "users", priority=0)
        q.submit(fetch, "posts", retries=2, retry_delay=0.25, timeout=5)
        summary = await q.run(strict=True)
        return summary.values


print(osiiso.run(main()))

---

Key Features

• Unified API — Same interface for async, threaded, and process queues
• Priority scheduling — Lower priority numbers execute first
• Retries with backoff — Configurable retry count, delay, and exponential backoff
• Timeouts — Per-task and queue-level time limits
• Graceful shutdown — must_complete tasks are protected during shutdown
• Batch workflows — submit(), map(), and group() for flexible task submission
• Structured results — RunSummary and immutable TaskResult records
• Lifecycle hooks — on_start, on_complete, and on_retry callbacks
• uvloop support — Optional acceleration through osiiso.run()
• Zero dependencies — No runtime dependencies; typed with py.typed

---

Quick Example

AsyncQueueThreadQueueProcessQueue

import asyncio
import osiiso

async def fetch(name: str) -> str:
    await asyncio.sleep(0.1)
    return f"fetched {name}"

async def main():
    async with osiiso.AsyncQueue(workers=4) as q:
        q.map(fetch, ["users", "posts", "comments"], retries=2, timeout=5)
        summary = await q.run(strict=True)
        print(summary.values)

osiiso.run(main())

import time
import osiiso

def resize(path: str) -> str:
    time.sleep(0.1)
    return f"resized {path}"

with osiiso.ThreadQueue(workers=4) as q:
    q.map(resize, ["a.png", "b.png", "c.png"], name="resize")
    summary = q.run(strict=True)

print(summary.values)

import osiiso

def score(n: int) -> int:
    return sum(i * i for i in range(n))

if __name__ == "__main__":
    with osiiso.ProcessQueue(workers=4) as q:
        q.map(score, [10_000, 20_000, 30_000], name="score")
        summary = q.run(strict=True)

    print(summary.values)

---

Architecture

graph LR
    A["submit() / map() / group()"] --> B["Priority Queue"]
    B --> C["Worker Pool"]
    C --> D["TaskHandle"]
    D --> E["TaskResult"]
    E --> F["RunSummary"]

    style A fill:#6366f1,color:#fff,stroke:none
    style B fill:#8b5cf6,color:#fff,stroke:none
    style C fill:#a855f7,color:#fff,stroke:none
    style D fill:#c084fc,color:#fff,stroke:none
    style E fill:#d8b4fe,color:#1e1b4b,stroke:none
    style F fill:#ede9fe,color:#1e1b4b,stroke:none

---

Next Steps

• Get Started

  ---

  Install osiiso and run your first task queue in under a minute.

  Quick Start

• User Guide

  ---

  Learn about task submission, options, handles, groups, and lifecycle policies.

  Task Submission

• API Reference

  ---

  Complete reference for every public class, method, and attribute.

  API Reference

• Examples

  ---

  See osiiso in action with the feature gallery and Hacker News pipeline.

  Examples