GitHub - aleksul/repid: Repid is a simple, fast, and extensible async task queue framework, with built-in AsyncAPI 3.0 schema generation · GitHub

main

Branches Tags

Go to file

Code

NameName

Last commit message

Last commit date

Latest commit History 666 Commits
.devcontainer		.devcontainer
.github		.github
docs		docs
repid		repid
tests		tests
.gitignore		.gitignore
.pre-commit-config.yaml		.pre-commit-config.yaml
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
SECURITY.md		SECURITY.md
mkdocs.yml		mkdocs.yml
pyproject.toml		pyproject.toml
uv.lock		uv.lock
View all files

Repository files navigation

repid

Repid is a simple, fast, and extensible async task queue framework, with built-in AsyncAPI 3.0 schema generation.

pip install repid

Features

AsyncAPI 3.0 out of the box - your schema is generated automatically as you build. No separate spec files to maintain, no drift between code and docs.
Broker flexibility - works with RabbitMQ, Redis, AWS SQS, Kafka, NATS, and Google Cloud Pub/Sub. Switch or run multiple brokers side by side without changing your actor code.
Extreme performance - built for maximum throughput and minimal overhead, making it the fastest async task queue in the Python ecosystem.
Developer ergonomics - define actors as plain async functions with dependency injection and Pydantic argument validation.
Built for testability - a drop-in TestClient lets you test actors in-memory without a running broker, making unit tests fast and dependency-free.
LLM-friendly documentation - grab llms.txt or llms-full.txt from the docs site to feed into your favorite AI coding assistant.

Quickstart

Here is how the simplest producer-consumer application looks, using AMQP server.

Producer:

import asyncio

from repid import AmqpServer, Repid

app = Repid(title="My App", version="1.0.0")
app.servers.register_server(
    "default",
    AmqpServer("amqp://user:password@localhost:5672"),
    is_default=True,
)


async def main() -> None:
    async with app.servers.default.connection():
        await app.send_message(
            channel="default",
            payload=b"",
            headers={"topic": "awesome_job"},
        )


asyncio.run(main())

Consumer:

import asyncio

from repid import AmqpServer, Repid, Router

app = Repid(title="My App", version="1.0.0")
app.servers.register_server(
    "default",
    AmqpServer("amqp://user:password@localhost:5672"),
    is_default=True,
)

router = Router()


@router.actor
async def awesome_job() -> None:
    print("Hello async jobs!")
    await asyncio.sleep(1.0)


app.include_router(router)


async def main() -> None:
    async with app.servers.default.connection():
        await app.run_worker()


asyncio.run(main())