mirror of
https://github.com/d3vyce/taskiq-deduplication.git
synced 2026-08-05 03:14:08 +00:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e75b9f3f20 | ||
|
|
56e82689f8 |
@@ -55,7 +55,7 @@ except DuplicateTaskError:
|
|||||||
- **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
|
- **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
|
||||||
- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
|
- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
|
||||||
- **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
|
- **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
|
||||||
- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
|
- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
|
||||||
- **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
|
- **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
|
||||||
- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
|
- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
|
||||||
|
|
||||||
|
|||||||
+1
-1
@@ -55,7 +55,7 @@ except DuplicateTaskError:
|
|||||||
- **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
|
- **Configurable TTL** — set a global default or override per task with the `deduplication_ttl` label.
|
||||||
- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
|
- **Lock heartbeat** — a background task re-extends the lock TTL while the task runs, so long-running tasks keep their lock instead of expiring mid-execution and admitting a duplicate.
|
||||||
- **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
|
- **Explicit lock key** — pin any task to a fixed Redis key with `deduplication_key`, bypassing fingerprint computation entirely.
|
||||||
- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments.
|
- **Partial fingerprint** — deduplicate on a subset of kwargs with `deduplication_key_fields`, ignoring irrelevant arguments (positional arguments are excluded).
|
||||||
- **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
|
- **Per-task opt-out** — disable deduplication for individual tasks with the `deduplication` label.
|
||||||
- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
|
- **Startup resilience** — automatic reconnection with exponential backoff if Redis is unavailable at broker startup.
|
||||||
|
|
||||||
|
|||||||
+14
-2
@@ -147,11 +147,19 @@ await my_task.kicker().with_labels(deduplication_ttl=60).kiq(user_id=42)
|
|||||||
| `deduplication` | `bool` | Set `False` to opt out of deduplication entirely for this task. |
|
| `deduplication` | `bool` | Set `False` to opt out of deduplication entirely for this task. |
|
||||||
| `deduplication_ttl` | `int` | Lock TTL in seconds. Overrides the middleware `default_ttl`. |
|
| `deduplication_ttl` | `int` | Lock TTL in seconds. Overrides the middleware `default_ttl`. |
|
||||||
| `deduplication_key` | `str` | Explicit lock key. Skips fingerprint computation entirely. |
|
| `deduplication_key` | `str` | Explicit lock key. Skips fingerprint computation entirely. |
|
||||||
| `deduplication_key_fields` | `list[str]` | Subset of kwargs to include in the fingerprint. Ignored if `deduplication_key` is set. |
|
| `deduplication_key_fields` | `list[str]` | Subset of kwargs to include in the fingerprint. Positional arguments are excluded. Ignored if `deduplication_key` is set. |
|
||||||
|
|
||||||
## Fingerprint and key customisation
|
## Fingerprint and key customisation
|
||||||
|
|
||||||
By default the lock key is a SHA-256 fingerprint of the task name and all kwargs.
|
By default the lock key is a SHA-256 fingerprint of the task name, its positional
|
||||||
|
arguments and all kwargs.
|
||||||
|
|
||||||
|
!!! warning "Positional and keyword calls fingerprint differently"
|
||||||
|
|
||||||
|
taskiq serialises arguments as they were passed, without binding them to the
|
||||||
|
task signature. `my_task.kiq(42)` and `my_task.kiq(user_id=42)` are therefore
|
||||||
|
*not* recognised as duplicates of each other. Call a deduplicated task
|
||||||
|
consistently, preferably always with keyword arguments.
|
||||||
|
|
||||||
### Explicit key
|
### Explicit key
|
||||||
|
|
||||||
@@ -182,6 +190,10 @@ If a listed field is absent from a task's kwargs, it is dropped from the
|
|||||||
fingerprint and a warning is logged, since this can make genuinely different
|
fingerprint and a warning is logged, since this can make genuinely different
|
||||||
calls collide on the same lock.
|
calls collide on the same lock.
|
||||||
|
|
||||||
|
Positional arguments are excluded from the fingerprint entirely when this label is
|
||||||
|
set: you asked to deduplicate on named fields, so pass them as keyword arguments.
|
||||||
|
A warning is logged if the task is called with positional arguments anyway.
|
||||||
|
|
||||||
## Opting out per task
|
## Opting out per task
|
||||||
|
|
||||||
```python
|
```python
|
||||||
|
|||||||
@@ -154,12 +154,21 @@ class RedisDeduplicationMiddleware(TaskiqMiddleware):
|
|||||||
message.task_name,
|
message.task_name,
|
||||||
missing,
|
missing,
|
||||||
)
|
)
|
||||||
|
if message.args:
|
||||||
|
logger.warning(
|
||||||
|
"Task %s was called with positional arguments but uses "
|
||||||
|
"deduplication_key_fields; positional arguments are excluded "
|
||||||
|
"from the fingerprint.",
|
||||||
|
message.task_name,
|
||||||
|
)
|
||||||
kwargs = {k: v for k, v in message.kwargs.items() if k in key_fields}
|
kwargs = {k: v for k, v in message.kwargs.items() if k in key_fields}
|
||||||
|
args: list[Any] = []
|
||||||
else:
|
else:
|
||||||
kwargs = message.kwargs
|
kwargs = message.kwargs
|
||||||
|
args = message.args
|
||||||
try:
|
try:
|
||||||
payload = json.dumps(
|
payload = json.dumps(
|
||||||
{"task": message.task_name, "kwargs": kwargs},
|
{"task": message.task_name, "args": args, "kwargs": kwargs},
|
||||||
sort_keys=True,
|
sort_keys=True,
|
||||||
)
|
)
|
||||||
except TypeError:
|
except TypeError:
|
||||||
|
|||||||
+4
-2
@@ -33,13 +33,15 @@ async def real_redis():
|
|||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def make_message():
|
def make_message():
|
||||||
def _make(task_name="my_task", task_id="task-1", labels=None, kwargs=None):
|
def _make(
|
||||||
|
task_name="my_task", task_id="task-1", labels=None, kwargs=None, args=None
|
||||||
|
):
|
||||||
return TaskiqMessage(
|
return TaskiqMessage(
|
||||||
task_id=task_id,
|
task_id=task_id,
|
||||||
task_name=task_name,
|
task_name=task_name,
|
||||||
labels=labels or {},
|
labels=labels or {},
|
||||||
labels_types={},
|
labels_types={},
|
||||||
args=[],
|
args=args or [],
|
||||||
kwargs=kwargs or {},
|
kwargs=kwargs or {},
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -86,6 +86,27 @@ class TestDefaultBuildDeduplicationKey:
|
|||||||
m1
|
m1
|
||||||
) == middleware._build_deduplication_key(m2)
|
) == middleware._build_deduplication_key(m2)
|
||||||
|
|
||||||
|
def test_different_args_different_key(self, middleware, make_message):
|
||||||
|
m1 = make_message(args=["a"])
|
||||||
|
m2 = make_message(args=["b"])
|
||||||
|
assert middleware._build_deduplication_key(
|
||||||
|
m1
|
||||||
|
) != middleware._build_deduplication_key(m2)
|
||||||
|
|
||||||
|
def test_arg_order_matters(self, middleware, make_message):
|
||||||
|
m1 = make_message(args=["a", "b"])
|
||||||
|
m2 = make_message(args=["b", "a"])
|
||||||
|
assert middleware._build_deduplication_key(
|
||||||
|
m1
|
||||||
|
) != middleware._build_deduplication_key(m2)
|
||||||
|
|
||||||
|
def test_key_fields_ignores_args(self, middleware, make_message):
|
||||||
|
m1 = make_message(args=["a"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
|
||||||
|
m2 = make_message(args=["b"], labels={DEDUP_KEY_FIELDS_LABEL: ["x"]})
|
||||||
|
assert middleware._build_deduplication_key(
|
||||||
|
m1
|
||||||
|
) == middleware._build_deduplication_key(m2)
|
||||||
|
|
||||||
def test_key_prefix_in_output(self, make_message):
|
def test_key_prefix_in_output(self, make_message):
|
||||||
mw = RedisDeduplicationMiddleware(
|
mw = RedisDeduplicationMiddleware(
|
||||||
redis_url="redis://localhost", key_prefix="myapp:locks"
|
redis_url="redis://localhost", key_prefix="myapp:locks"
|
||||||
|
|||||||
@@ -1266,15 +1266,15 @@ wheels = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "pymdown-extensions"
|
name = "pymdown-extensions"
|
||||||
version = "10.21.3"
|
version = "11.0"
|
||||||
source = { registry = "https://pypi.org/simple" }
|
source = { registry = "https://pypi.org/simple" }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "markdown" },
|
{ name = "markdown" },
|
||||||
{ name = "pyyaml" },
|
{ name = "pyyaml" },
|
||||||
]
|
]
|
||||||
sdist = { url = "https://files.pythonhosted.org/packages/9e/26/d1015444da4d952a1ca487a236b522eb979766f0295a0bd0c5fc089989a9/pymdown_extensions-10.21.3.tar.gz", hash = "sha256:72cfcf55f07aea0d4af2c4f11dd4e52466ddfb1bb819673146398e0bd3a77354", size = 854140, upload-time = "2026-05-13T12:57:32.267Z" }
|
sdist = { url = "https://files.pythonhosted.org/packages/47/67/f1e79672a5f91985577c7984c9709ca110e4fd37fe7fd167b60422e6ccc2/pymdown_extensions-11.0.tar.gz", hash = "sha256:8269cef0247f9e2d0a62fcea10860aba05c1cbab5470fd4b63230b96434dc589", size = 857049, upload-time = "2026-06-23T02:27:45.146Z" }
|
||||||
wheels = [
|
wheels = [
|
||||||
{ url = "https://files.pythonhosted.org/packages/7e/85/545a951eecc270fcd688288c600017e2050a1aacb56c711d208586d3e470/pymdown_extensions-10.21.3-py3-none-any.whl", hash = "sha256:d7a5d08014fc571e80ca21dd6f854e31f94c489800350564d55d15b3c41e76b6", size = 269002, upload-time = "2026-05-13T12:57:30.296Z" },
|
{ url = "https://files.pythonhosted.org/packages/af/b6/1ae53367e28b9cffa3be7574e13fbe4589694272fd47710fbdbafd3d63c6/pymdown_extensions-11.0-py3-none-any.whl", hash = "sha256:fbc4acb641814fa9d17521bbd21a5240ef739a662f11c06330c4b78c93e954d6", size = 269415, upload-time = "2026-06-23T02:27:43.826Z" },
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
|
|||||||
Reference in New Issue
Block a user