Struct SqliteStorage 

pub struct SqliteStorage<Args> { /* private fields */ }

SqliteStorage is a storage backend for apalis using sqlite as the database.

It supports both standard polling and event-driven (hooked) storage mechanisms.

Features

Feature	Status	Description
Backend	✅	Supports storage and retrieval of tasks
TaskSink	✅	Ability to push new tasks
Serialization	✅	Serialization support for arguments
Workflow	✅	Flexible enough to support workflows
Web Interface	✅	Expose a web interface for monitoring tasks
FetchById	✅	Allow fetching a task by its ID
RegisterWorker	✅	Allow registering a worker with the backend
BackendFactory	✅	Share one connection across multiple workers via SqliteStorageFactory
WaitForCompletion	✅	Wait for tasks to complete without blocking
ResumeById	✅	Resume a task by its ID
ResumeAbandoned	✅	Resume abandoned tasks
ListWorkers	✅	List all workers registered with the backend
ListTasks	✅	List all tasks in the backend

Key: ✅ : Supported | ⚠️ : Not implemented | ❌ : Not Supported | ❗ Limited support

Tests:

Backend

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    async fn task(task: u32, worker: WorkerContext) {
        // Do some work 
    }
    let worker = WorkerBuilder::new("backend-test")
        .backend(backend)
        .build(task);
    let _ = worker.stream().take(1).collect::<Vec<_>>().await;
}

TaskSink

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push(42).await.unwrap();

    async fn task(task: u32, worker: WorkerContext) {
        worker.stop().unwrap();
    }
    let worker = WorkerBuilder::new("task-sink-test")
        .backend(backend)
        .build(task);
    worker.run().await.unwrap();
}

Serialization

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
   fn assert_codec<B: BackendConfig + WireFormatBackend>(backend: B) 
   where
       B::Codec: Codec<B::Args, Compact=Vec<u8>>,
   {
   }
   assert_codec(backend);
}

Workflow

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    backend.push(42).await.unwrap();

    async fn task1(task: u32, worker: WorkerContext) -> u32 {
        task + 99 
    }
    async fn task2(task: u32, worker: WorkerContext) -> u32 {
        task + 1 
    }
    async fn task3(task: u32, worker: WorkerContext) {
        assert_eq!(task, 142);
        worker.stop().unwrap();
    }
    let workflow = SteppedFlow::new("test-workflow")
       .and_then(task1)
       .and_then(task2)
       .and_then(task3);
    let worker = WorkerBuilder::new("workflow-test")
        .backend(backend)
        .build(workflow);
    worker.run().await.unwrap();
}

WebUI

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_web_ui<B: Expose<u32, Durable>>(backend: B) {};
    assert_web_ui(backend);
}

WaitForCompletion

#[tokio::main]
async fn main() {
    // let mut backend = /* snip */;
    
    fn assert_wait_for_completion<B: WaitForCompletion<u32> + Backend>(backend: B) {};
    assert_wait_for_completion(backend);
}

Implementations

impl SqliteStorage<()>

pub fn connect_with_callback( url: &str, ) -> Result<(SqlitePool, HookCallbackListener), Error>

Connects to a database returning a pool and listener

pub async fn setup(pool: &SqlitePool) -> Result<(), Error>

Perform migrations for storage

pub fn migrations() -> Migrator

Get sqlite migrations without running them

impl<T> SqliteStorage<T>

pub fn new(pool: &SqlitePool) -> SqliteStorage<T>

Create a new SqliteStorage

pub fn with_config(self, config: Config) -> SqliteStorage<T>

Create a new SqliteStorage with a custom configuration

pub fn with_callback( self, callback: HookCallbackListener, ) -> PollWith<Self, StreamStrategy<HookCallbackListener>>

Attach a callback to an instance

pub fn pool(&self) -> &SqlitePool

Get the underlying pool

Trait Implementations

impl<Args> Backend for SqliteStorage<Args>

type Task = Task<Vec<u8>>

The type of task the backend emits.

type Error = Error

The error type returned by backend operations

fn poll_ready( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Result<(), Self::Error>>

Polls whether the backend is ready for the worker to request more work.

fn poll_next( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Option<Result<SqliteTask, Self::Error>>>

Polls the backend for the next available task for this worker.

fn poll_close( &mut self, cx: &mut Context<'_>, worker: &WorkerContext, ) -> Poll<Result<(), Self::Error>>

Flushes/releases any resources the backend holds (pending acks, open subscriptions, connections) before the worker fully shuts down.

impl<Args> BackendConfig for SqliteStorage<Args>

type Args = Args

The type of argument this backend emits

type Id = Ulid

The internal type of this Backend’s TaskId

type Kind = Durable

This defines the kind of Backend

type Config = Config

The config for the backend

type Layer = TaskPersistLayer<JsonCodec<Value>, Value>

The type representing backend middleware layer.

fn config(&self) -> &Self::Config

Returns the config associated with the backend.

fn middleware(&mut self, _: &mut WorkerContext) -> Self::Layer

Returns the backend’s middleware layer.

impl<T> Clone for SqliteStorage<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<Args: Debug> Debug for SqliteStorage<Args>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<Args> FetchById for SqliteStorage<Args>

where Self: Backend<Error = Error, Task = Task<Vec<u8>>>, Args: 'static,

fn fetch_by_id( &mut self, id: &TaskId, ) -> impl Future<Output = Result<Option<SqliteTask>, Self::Error>> + Send

Fetch a task by its unique identifier

impl<Args> ListAllTasks for SqliteStorage<Args>

where Self: Backend<Error = Error>,

fn list_all_tasks( &self, filter: &Filter, ) -> impl Future<Output = Result<Vec<SqliteTask>, Self::Error>> + Send

List tasks matching the given filter in all queues

impl<Args> ListQueues for SqliteStorage<Args>

where Self: Backend<Error = Error>,

fn list_queues( &self, ) -> impl Future<Output = Result<Vec<QueueInfo>, Self::Error>> + Send

List all available queues in the backend

impl<Args> ListTasks for SqliteStorage<Args>

where Self: Backend<Error = Error>, Args: 'static,

fn list_tasks( &self, filter: &Filter, ) -> impl Future<Output = Result<Vec<SqliteTask>, Self::Error>> + Send

List tasks matching the given filter in the current queue

impl<Args: Sync> ListWorkers for SqliteStorage<Args>

where Self: Backend<Error = Error>,

fn list_workers( &self, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in the current queue

fn list_all_workers( &self, ) -> impl Future<Output = Result<Vec<RunningWorker>, Self::Error>> + Send

List all registered workers in all queues

impl<Args> Metrics for SqliteStorage<Args>

where Self: Backend<Error = Error>,

fn global( &self, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Collects and returns global statistics from the backend

fn fetch_by_queue( &self, ) -> impl Future<Output = Result<Vec<Statistic>, Self::Error>> + Send

Collects and returns statistics for a specific queue

impl<Args> Sink<Task<Vec<u8>>> for SqliteStorage<Args>

type Error = Error

The type of value produced by the sink when an error occurs.

fn poll_ready( self: Pin<&mut Self>, _: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Attempts to prepare the Sink to receive a value.

fn start_send(self: Pin<&mut Self>, item: SqliteTask) -> Result<(), Self::Error>

Begin the process of sending a value to the sink. Each call to this function must be preceded by a successful call to poll_ready which returned Poll::Ready(Ok(())).

fn poll_flush( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output from this sink.

fn poll_close( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Result<(), Self::Error>>

Flush any remaining output and close this sink, if necessary.

impl<T> TryNewBackend for SqliteStorage<T>

type Backend = BeforeStart<SqliteStorage<T>, <SqliteStorage<T> as Backend>::Error>

The final backend after decoration

fn try_new(config: Self::Config) -> Result<Self::Backend, Self::Error>

Attempt to construct a new backend instance from its configuration.

impl<'pin, Args> Unpin for SqliteStorage<Args>

where PinnedFieldsOf<__SqliteStorage<'pin, Args>>: Unpin,

impl<Args> Vacuum for SqliteStorage<Args>

where Self: Backend<Error = Error>, Args: Send,

async fn vacuum(&mut self) -> Result<usize, Self::Error>

Cleans up resources and returns the number of items vacuumed

impl<Args, O> WaitForCompletion<O> for SqliteStorage<Args>

where Self: Backend<Error = Error>, O: Send + 'static + DeserializeOwned,

type ResultStream = Pin<Box<dyn Stream<Item = Result<TaskResult<O>, <SqliteStorage<Args> as Backend>::Error>> + Send>>

The result stream type yielding task results

fn wait_for( &self, task_ids: impl IntoIterator<Item = TaskId>, ) -> Self::ResultStream

Wait for multiple tasks to complete, yielding results as they become available

fn check_status( &self, task_ids: impl IntoIterator<Item = TaskId> + Send, ) -> impl Future<Output = Result<Vec<TaskResult<O>>, Self::Error>> + Send

Check current status of tasks without waiting

fn wait_for_single(&self, task_id: TaskId) -> Self::ResultStream

Wait for a single task to complete, yielding its result

impl<Args> WireFormatBackend for SqliteStorage<Args>

type Codec = JsonCodec

The codec used to encode and decode tasks.

type Compact = Vec<u8>

The compact representation of task arguments.

fn codec(&self) -> &Self::Codec

Returns a reference to the codec used by the backend.