Skip to content

API Reference

You can import them directly from taskiq_deduplication:

from taskiq_deduplication import (
    DuplicateTaskError,
    RedisDeduplicationMiddleware,
)

taskiq_deduplication.RedisDeduplicationMiddleware

Bases: TaskiqMiddleware

Prevents duplicate tasks from being queued.

When a task is dispatched, a Redis lock is acquired for the duration of its execution. Any subsequent task with the same fingerprint is rejected with DuplicateTaskError while the lock is held. The lock is released automatically on completion or error.

Attributes:

Name Type Description
redis_url

Redis connection URL (str or RedisDsn) passed to Redis.from_url.
default_deduplication

Whether deduplication is enabled by default.
default_ttl

Default lock TTL in seconds.
key_prefix

Prefix for all Redis lock keys.
heartbeat

Whether to periodically re-extend the lock TTL during task execution so long-running tasks keep their lock.
heartbeat_interval

Seconds between heartbeat refreshes. When None it defaults to a third of the task's TTL (with a 1s floor).
fail_open

Whether a Redis error while acquiring the lock lets the task through instead of aborting the send. Detected duplicates still raise DuplicateTaskError.

taskiq_deduplication.DuplicateTaskError

Bases: Exception

Raised when a task with identical name and kwargs is already queued or running.

Attributes:

Name Type Description
task_name

Name of the task that was rejected.
key

Redis lock key whose owner caused the rejection.
holder_task_id

task_id of the task currently holding the lock, or None if it could not be retrieved (e.g. the lock was released between the failed acquisition and the lookup).