persist-queue

Compatibility

• Storage Format: Async queues are compatible with synchronous queues using the same storage format.
• API Design: The interface is designed to be consistent with the synchronous API.
• Error Handling: Uses the same exception types (e.g., Empty, Full).
• Serialization: Supports the same serializers used in the sync version.

Important Limitations

• Serializers: The current implementation uses synchronous serializers. If you are serializing very large amounts of data, the serialization process may block the event loop.
• Python Version: Requires Python 3.7+ to support the async with context manager implementation.

The library provides several asynchronous queue implementations depending on your storage and ordering requirements:

• AsyncQueue: An asynchronous file-based queue, optimized for high-throughput scenarios.
• AsyncSQLiteQueue: An asynchronous SQLite-based queue, providing better transaction support and query capabilities. This is an alias for AsyncFIFOSQLiteQueue.
• AsyncFIFOSQLiteQueue: A First-In-First-Out SQLite queue (alias for AsyncSQLiteQueue).
• AsyncFILOSQLiteQueue: A Last-In-First-Out (LIFO) SQLite queue.
• AsyncUniqueQ: A SQLite queue that ensures no duplicate items are stored.

To get the best performance out of persist-queue, consider the following tips:

• SQLite3 Queues: Use WAL mode (enabled by default) for better performance. For batch operations, set auto_commit=False and call task_done() to persist changes.
• MySQL Queues: These use connection pooling for better performance.
• Protocol Selection: The library automatically selects the optimal pickle protocol for you.
• Windows: Performance for the File queue has been significantly improved in recent versions (v0.4.1+).

Standard installation may not include all features. Use the following commands to enable specific capabilities:

• For async support (requires Python 3.7+): pip install "persist-queue[async]"
• For MySQL support: pip install "persist-queue[extra]"

When moving from the synchronous API to the asynchronous API, follow these steps:

1. Replace Queue with AsyncQueue.
2. Replace SQLiteQueue with AsyncSQLiteQueue.
3. Use async with context managers for queue lifecycle.
4. Add the await keyword before all queue operations.
5. Ensure all queue operations are called within an async function.

Note: Async queues use the same storage format as sync queues and are compatible (they can read each other's data).

# Sync version
from persistqueue import Queue
queue = Queue("/path/to/queue")
queue.put("data")
item = queue.get()
queue.task_done()

# Async version
from persistqueue import AsyncQueue
async with AsyncQueue("/path/to/queue") as queue:
    await queue.put("data")
    item = await queue.get()
    await queue.task_done()

You can benchmark the performance of various queue types (including async) using the built-in benchmarking tool. This allows you to compare sync and async queues on your specific platform.

To run benchmarks directly, use the benchmark/run_benchmark.py script. The first argument specifies the number of items to test (default is 1000), and the second argument specifies the output format: rst (for a reStructuredText table), console, or json.

python benchmark/run_benchmark.py 1000 rst

For asynchronous environments (Python 3.7+), use AsyncQueue (file-based) or AsyncSQLiteQueue (SQLite-based). These should be used with async with to ensure proper lifecycle management.

Async File Queue

import asyncio
from persistqueue import AsyncQueue

async def main():
    async with AsyncQueue("/path/to/queue") as queue:
        await queue.put("async item")
        item = await queue.get()
        await queue.task_done()

asyncio.run(main())

Async SQLite Queue

import asyncio
from persistqueue import AsyncSQLiteQueue

async def main():
    async with AsyncSQLiteQueue("/path/to/queue.db") as queue:
        item_id = await queue.put({"key": "value"})
        item = await queue.get()
        await queue.update({"key": "new_value"}, item_id)
        await queue.task_done()

asyncio.run(main())

To use the asynchronous persistent queue API, you must install the aiofiles and aiosqlite packages. You can install them individually or via the requirements file.

pip install aiofiles aiosqlite
# OR
pip install -r requirements.txt

To use the asynchronous queue implementations, you must install the async extra. This ensures that necessary dependencies like aiofiles and aiosqlite are included.

Recommended installation:

pip install "persist-queue[async]"

Manual dependency installation: If you prefer to manage dependencies manually, ensure you have:

• aiofiles>=0.8.0
• aiosqlite>=0.17.0

You can install persist-queue using pip. Depending on your requirements, you may need to install extra dependencies for specific features like MySQL or async support.

Basic Installation

pip install persist-queue

Installation with Extra Features

• msgpack, cbor, and MySQL support: pip install "persist-queue[extra]"
• Async support (requires Python 3.7+): pip install "persist-queue[async]"
• All features: pip install "persist-queue[extra,async]"

Requirements

• Python 3.5 or newer.
• For async features: Python 3.7+ with aiofiles and aiosqlite.
• For MySQL queues: DBUtils and PyMySQL.

File-based Queue

Use Queue for a basic file-based FIFO queue. It uses pickle by default for serialization.

from persistqueue import Queue

q = Queue("my_queue_path")
q.put("item1")
item = q.get()
q.task_done()

SQLite-based Queue

Use SQLiteQueue or FIFOSQLiteQueue for SQLite-backed storage. You can enable auto_commit=True for immediate persistence.

import persistqueue

q = persistqueue.SQLiteQueue('my_queue.db', auto_commit=True)
q.put('data1')
item = q.get()