Crate apalis

Source

Expand description

apalis

Simple and extensible multithreaded background task and messages processing library for Rust

§Features

Simple, predictable task handling - Task handlers are just async functions, with a macro-free API
Robust task execution - Built-in retries, rate limiting, timeouts, idempotency, and error handling
Multiple storage backends - Redis, PostgreSQL, SQLite, and in-memory storage
Advanced task management - Prioritization, scheduling, metadata, and result tracking
Scalable by design - Distributed backends with configurable concurrency and multi-threaded execution
Lightweight and lazy - Like a Future, workers do nothing until polled or woken: no busy-waiting
Familiar dependency injection - Similar to popular frameworks like actix and axum
Runtime agnostic - Works with tokio, async-std, and other async runtimes
Production ready - Built-in monitoring, metrics, graceful shutdown, and comprehensive error reporting
Extensible middleware - Take full advantage of the tower ecosystem of services and utilities
Optional web interface - Manage and monitor tasks through a web UI

§Crate ecosystem

§Message queues

Crate	Examples	Features
`apalis-amqp`		🔁 📅 🔒 📡 👷
`apalis-nats`		🔁 📅 🔒 📡 👷
`apalis-rsmq`		🔁 📅 🔒 📡 👷

§Persistence backends

Crate	Examples	Features
`apalis-redis`		🔁 📅 🔒 📡 👷 ⏯️
`apalis-sqlite`		🔁 📅 🔒 📡 👷 ⏯️
`apalis-postgres`		🔁 📅 🔒 📡 👷 ⏯️
`apalis-mysql`		🔁 📅 🔒 📡 👷 ⏯️
`apalis-pgmq`		🔁 📅 🔒 📡 👷 ⏯️

§Scheduling

Crate	Examples	Features
`apalis-cron`		🔁 📅

§Core & tooling

Crate	Examples	Features
`apalis`		🔁 📅 🔒 📡 👷
`apalis-core`	:	🔁 📅
`apalis-workflow`		🔁 📅 🔒
`apalis-board`		🔁 📅 🔒 📡 👷
`apalis-board-api`		🔁 📅 🔒 📡 👷

§Feature legend

🔁 Retries: automatically retry failed tasks.
📅 Scheduling: delayed or scheduled execution.
🔒 Unique jobs: deduplication or idempotency.
⏯️ Resumable jobs: resume interrupted jobs or workflows.
🔐 Task leases: time-limited task ownership, visibility timeouts, or orphan recovery.
📡 Pubsub / realtime: event-driven notifications or realtime updates.
👷 Worker management: worker discovery, monitoring, lifecycle, or management APIs.

§Getting Started

To get started, just add to Cargo.toml

[dependencies]
apalis = { version = "1.0.0" }
# apalis-redis = { version = "1.0.0" } # Use redis/sqlite/postgres etc

§Usage

use apalis::prelude::*;
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Serialize, Clone)]
struct Email {
    to: String,
}

/// A function called for every task
async fn send_email(task: Email, data: Data<usize>) {
  // execute task
}


#[tokio::main]
async fn main() {
    let mut storage = MemoryStorage::new();
     storage
        .push(Email {
            to: "test@example.com".to_string(),
        })
        .await
        .expect("Could not add tasks");

    WorkerBuilder::new("email-worker")
      .backend(storage)
      .concurrency(2)
      .parallelize(tokio::spawn)
      .data(0usize)
      .build(send_email)
      .run()
      .await;
}

§Workflows

apalis has first class support for sequential, dag and conditional workflows.

use apalis::prelude::*;
use apalis_workflow::*;
use std::time::Duration;
use apalis_file_storage::JsonStorage;;

#[tokio::main]
async fn main() {
   let workflow = SteppedFlow::new("odd-numbers-workflow")
       .delay_for(Duration::from_millis(1000))
       .and_then(|a: usize| async move { Ok::<_, BoxDynError>((0..a).collect::<Vec<_>>()) })
       .filter_map(|x| async move { if x % 2 != 0 { Some(x) } else { None } })
       .and_then(|a: Vec<usize>| async move {
           println!("Sum: {}", a.iter().sum::<usize>());
           Ok::<_, BoxDynError>(())
        });

   let mut in_memory = JsonStorage::new_temp().unwrap();

   in_memory.push_start(10).await.unwrap();

   let worker = WorkerBuilder::new("rango-tango")
       .backend(in_memory)
       .on_event(|ctx, ev| {
           println!("On Event = {:?}", ev);
       })
       .build(workflow);
   worker.run().await.unwrap();
}

For more functionality like fold, filter_map and other combinators checkout the docs

§How apalis works

Here is a basic example of how the core parts integrate

sequenceDiagram
    participant App
    participant Worker
    participant Backend

    App->>+Backend: Add task to queue
    Backend-->>+Worker: Job data
    Worker->>+Backend: Update task status to 'Running'
    Worker->>+App: Started task
    loop task execution
        Worker-->>-App: Report task progress
    end
    Worker->>+Backend: Update task status to 'completed'

§Observability

With the web UI, you can manage your jobs through a simple interface. Check out this working example to see how it works.

Workers Screenshot

§Integrations

zino: Next-generation framework for composable applications in Rust.
spring-rs: Application framework written in Rust, inspired by Java’s SpringBoot

§Projects using apalis

Ryot: A self hosted platform for tracking various facets of your life - media, fitness etc.
Hyprnote: Local-first AI Notepad for Private Meetings
oxy: A framework for building SQL agents and automations for analytics.
OpenZeppelin Relayer: OpenZeppelin relayer service
Summarizer: Podcast summarizer
Universal Inbox: Universal Inbox is a solution that centralizes all your notifications and tasks in one place to create a unique inbox.
Hatsu: Self-hosted and fully-automated ActivityPub bridge for static sites.
Stamon: A lightweight self-hosted status monitoring tool
Gq: open-source filtering tool for JSON and YAML files

§External examples

Shuttle: Using apalis-cron with shuttle.rs
Actix-Web: Using apalis-redis with actix-web
Zino-Example: Using zino

§Resources

§Thanks to

tower - Tower is a library of modular and reusable components for building robust networking clients and servers.
redis-rs - Redis library for rust
sqlx - The Rust SQL Toolkit
cron - A cron expression parser and schedule explorer

§Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests to us.

§Versioning

We use SemVer for versioning. For the versions available, see the tags on this repository.

§Contributors

See also the list of contributors who participated in this project.

§License

This project is licensed under the MIT License - see the LICENSE.md file for details

§Feature flags

tracing (enabled by default) — Support Tracing 👀
sentry — Support for Sentry exception and performance monitoring
prometheus — Support Prometheus metrics
opentelemetry — Support OpenTelemetry metrics
retry (enabled by default) — Support direct retrying tasks
timeout (enabled by default) — Support timeouts on tasks
limit (enabled by default) — 💪 Limit the amount of tasks
filter — Support filtering tasks based on a predicate
catch-panic (enabled by default) — Captures panics in executions and convert them to errors
config (enabled by default) — Allow running workers using configuration
task-context — Enables tracking of tasks
shared — Enables cloning backends that dont implement clone
full — All available features

Modules§

configconfig: Worker configuration utilities allowing easier worker decoration
layers: Inbuilt middleware build on top of tower’s Layer
prelude: A “prelude” of common imports

Crate apalis

Crate apalis Copy item path

apalis

§Features

§Crate ecosystem

§Message queues

§Persistence backends

§Scheduling

§Core & tooling

§Feature legend

§Getting Started

§Usage

§Workflows

§How apalis works

§Observability

§Integrations

§Projects using apalis

§External examples

§Resources

§Thanks to

§Contributing

§Versioning

§Contributors

§License

§Feature flags

Modules§

Crate apalis