Skip to content

Commit 8ded366

Browse files
danfimovdanfimov
authored
Merge pull request #135 from taskiq-python/fix-134-support-for-taskiq-0-13-0
fix: compatibility with taskiq 0.13.0
2 parents 39c6b7c + b59493a commit 8ded366

5 files changed

Lines changed: 293 additions & 225 deletions

File tree

‎pyproject.toml‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@ classifiers = [
2727
]
2828
dependencies = [
2929
"redis>=8.0.0,<9",
30-
"taskiq>=0.12.0",
30+
"taskiq>=0.13.0",
3131
]
3232

3333
[project.urls]
@@ -42,8 +42,8 @@ dev = [
4242
]
4343
lint = [
4444
"black>=25.11.0",
45-
"mypy>=2.0.0",
46-
"ruff>=0.14.7",
45+
"mypy>=2.3.1",
46+
"ruff>=0.16.9",
4747
]
4848
test = [
4949
"fakeredis>=2.32.1",
@@ -112,6 +112,7 @@ lint.select = [
112112
"RUF", # Specific to Ruff checks
113113
"FA102", # Future annotations
114114
"UP", # Pyupgrade
115+
"LOG", # Logging
115116
]
116117
lint.ignore = [
117118
"D105", # Missing docstring in magic method
@@ -123,6 +124,7 @@ lint.ignore = [
123124
"ANN401", # typing.Any are disallowed in `**kwargs
124125
"PLR0913", # Too many arguments for function call
125126
"D106", # Missing docstring in public nested class
127+
"PLR0917", # Too many positional arguments
126128
]
127129
exclude = [".venv/"]
128130
line-length = 88

‎taskiq_redis/list_schedule_source.py‎

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,6 @@
55
from redis.asyncio import BlockingConnectionPool, Redis
66
from taskiq import ScheduledTask, ScheduleSource
77
from taskiq.abc.serializer import TaskiqSerializer
8-
from taskiq.compat import model_dump, model_validate
98
from taskiq.serializers import PickleSerializer
109
from typing_extensions import Self
1110

@@ -64,7 +63,7 @@ async def startup(self) -> None:
6463
logger.info("Migrating schedules from previous source")
6564
await self._previous_schedule_source.startup()
6665
schedules = await self._previous_schedule_source.get_schedules()
67-
logger.info(f"Found {len(schedules)}")
66+
logger.info("Found %d", len(schedules))
6867
for schedule in schedules:
6968
await self.add_schedule(schedule)
7069
if self._delete_schedules_after_migration:
@@ -143,8 +142,7 @@ async def delete_schedule(self, schedule_id: str) -> None:
143142
raw_schedule = await redis.getdel(self._get_data_key(schedule_id))
144143
if raw_schedule is not None:
145144
logger.debug("Deleting schedule %s", schedule_id)
146-
schedule = model_validate(
147-
ScheduledTask,
145+
schedule = ScheduledTask.model_validate(
148146
self._serializer.loadb(raw_schedule), # type: ignore[arg-type]
149147
)
150148
# We need to remove the schedule from the cron or time list.
@@ -162,7 +160,7 @@ async def add_schedule(self, schedule: "ScheduledTask") -> None:
162160
# At first we set data key which contains the schedule data.
163161
await redis.set(
164162
f"{self._prefix}:data:{schedule.schedule_id}",
165-
self._serializer.dumpb(model_dump(schedule)),
163+
self._serializer.dumpb(schedule.model_dump(mode="json")),
166164
)
167165
# Then we add the schedule to the cron or time list.
168166
# This is an optimization, so we can get all the schedules
@@ -229,7 +227,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
229227
buffer = buffer[self._buffer_size :]
230228

231229
return [
232-
model_validate(ScheduledTask, self._serializer.loadb(schedule)) # type: ignore[arg-type]
230+
ScheduledTask.model_validate(self._serializer.loadb(schedule)) # type: ignore[arg-type]
233231
for schedule in schedules
234232
if schedule
235233
]

‎taskiq_redis/redis_backend.py‎

Lines changed: 14 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@
77
from redis.asyncio.connection import Connection
88
from taskiq import AsyncResultBackend
99
from taskiq.abc.serializer import TaskiqSerializer
10-
from taskiq.compat import model_dump, model_validate
1110
from taskiq.depends.progress_tracker import TaskProgress
1211
from taskiq.result import TaskiqResult
1312
from taskiq.serializers import PickleSerializer
@@ -107,7 +106,7 @@ async def set_result(
107106
:param result: TaskiqResult instance.
108107
"""
109108
name = self._task_name(task_id)
110-
value = self.serializer.dumpb(model_dump(result))
109+
value = self.serializer.dumpb(result.model_dump(mode="json"))
111110
async with Redis(connection_pool=self.redis_pool) as redis:
112111
if self.result_ex_time:
113112
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -154,8 +153,7 @@ async def get_result(
154153
if result_value is None:
155154
raise ResultIsMissingError
156155

157-
taskiq_result = model_validate(
158-
TaskiqResult[_ReturnType],
156+
taskiq_result = TaskiqResult[_ReturnType].model_validate(
159157
self.serializer.loadb(result_value), # type: ignore[arg-type]
160158
)
161159

@@ -179,7 +177,7 @@ async def set_progress(
179177
:param result: task's TaskProgress instance.
180178
"""
181179
name = self._task_name(task_id) + PROGRESS_KEY_SUFFIX
182-
value = self.serializer.dumpb(model_dump(progress))
180+
value = self.serializer.dumpb(progress.model_dump(mode="json"))
183181
async with Redis(connection_pool=self.redis_pool) as redis:
184182
if self.result_ex_time:
185183
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -206,8 +204,7 @@ async def get_progress(
206204
if result_value is None:
207205
return None
208206

209-
return model_validate(
210-
TaskProgress[_ReturnType],
207+
return TaskProgress[_ReturnType].model_validate(
211208
self.serializer.loadb(result_value), # type: ignore[arg-type]
212209
)
213210

@@ -286,7 +283,7 @@ async def set_result(
286283
:param result: TaskiqResult instance.
287284
"""
288285
name = self._task_name(task_id)
289-
value = self.serializer.dumpb(model_dump(result))
286+
value = self.serializer.dumpb(result.model_dump(mode="json"))
290287
async with self.redis as redis:
291288
if self.result_ex_time:
292289
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -331,8 +328,9 @@ async def get_result(
331328
if result_value is None:
332329
raise ResultIsMissingError
333330

334-
taskiq_result: TaskiqResult[_ReturnType] = model_validate(
335-
TaskiqResult[_ReturnType],
331+
taskiq_result: TaskiqResult[_ReturnType] = TaskiqResult[
332+
_ReturnType
333+
].model_validate(
336334
self.serializer.loadb(result_value), # type: ignore[arg-type]
337335
)
338336

@@ -356,7 +354,7 @@ async def set_progress(
356354
:param result: task's TaskProgress instance.
357355
"""
358356
name = self._task_name(task_id) + PROGRESS_KEY_SUFFIX
359-
value = self.serializer.dumpb(model_dump(progress))
357+
value = self.serializer.dumpb(progress.model_dump(mode="json"))
360358
async with self.redis as redis:
361359
if self.result_ex_time:
362360
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -382,8 +380,7 @@ async def get_progress(
382380
if result_value is None:
383381
return None
384382

385-
return model_validate(
386-
TaskProgress[_ReturnType],
383+
return TaskProgress[_ReturnType].model_validate(
387384
self.serializer.loadb(result_value), # type: ignore[arg-type]
388385
)
389386

@@ -470,7 +467,7 @@ async def set_result(
470467
:param result: TaskiqResult instance.
471468
"""
472469
name = self._task_name(task_id)
473-
value = self.serializer.dumpb(model_dump(result))
470+
value = self.serializer.dumpb(result.model_dump(mode="json"))
474471
async with self._acquire_master_conn() as redis:
475472
if self.result_ex_time:
476473
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -517,8 +514,7 @@ async def get_result(
517514
if result_value is None:
518515
raise ResultIsMissingError
519516

520-
taskiq_result = model_validate(
521-
TaskiqResult[_ReturnType],
517+
taskiq_result = TaskiqResult[_ReturnType].model_validate(
522518
self.serializer.loadb(result_value), # type: ignore[arg-type]
523519
)
524520

@@ -542,7 +538,7 @@ async def set_progress(
542538
:param result: task's TaskProgress instance.
543539
"""
544540
name = self._task_name(task_id) + PROGRESS_KEY_SUFFIX
545-
value = self.serializer.dumpb(model_dump(progress))
541+
value = self.serializer.dumpb(progress.model_dump(mode="json"))
546542
async with self._acquire_master_conn() as redis:
547543
if self.result_ex_time:
548544
await redis.set(name=name, value=value, ex=self.result_ex_time)
@@ -569,8 +565,7 @@ async def get_progress(
569565
if result_value is None:
570566
return None
571567

572-
return model_validate(
573-
TaskProgress[_ReturnType],
568+
return TaskProgress[_ReturnType].model_validate(
574569
self.serializer.loadb(result_value), # type: ignore[arg-type]
575570
)
576571

‎taskiq_redis/schedule_source.py‎

Lines changed: 7 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,6 @@
1212
)
1313
from taskiq import ScheduleSource
1414
from taskiq.abc.serializer import TaskiqSerializer
15-
from taskiq.compat import model_dump, model_validate
1615
from taskiq.scheduler.scheduled_task import ScheduledTask
1716
from taskiq.serializers import PickleSerializer
1817

@@ -50,8 +49,7 @@ def __init__(
5049
**connection_kwargs: Any,
5150
) -> None:
5251
warnings.warn(
53-
"RedisScheduleSource is deprecated. "
54-
"Please switch to ListRedisScheduleSource",
52+
"RedisScheduleSource is deprecated. Please switch to ListRedisScheduleSource", # noqa: E501
5553
DeprecationWarning,
5654
stacklevel=2,
5755
)
@@ -81,7 +79,7 @@ async def add_schedule(self, schedule: ScheduledTask) -> None:
8179
async with Redis(connection_pool=self.connection_pool) as redis:
8280
await redis.set(
8381
f"{self.prefix}:{schedule.schedule_id}",
84-
self.serializer.dumpb(model_dump(schedule)),
82+
self.serializer.dumpb(schedule.model_dump(mode="json")),
8583
)
8684

8785
async def get_schedules(self) -> list[ScheduledTask]:
@@ -103,7 +101,7 @@ async def get_schedules(self) -> list[ScheduledTask]:
103101
if buffer:
104102
schedules.extend(await redis.mget(buffer))
105103
return [
106-
model_validate(ScheduledTask, self.serializer.loadb(schedule)) # type: ignore[arg-type]
104+
ScheduledTask.model_validate(self.serializer.loadb(schedule)) # type: ignore[arg-type]
107105
for schedule in schedules
108106
if schedule
109107
]
@@ -163,7 +161,7 @@ async def add_schedule(self, schedule: ScheduledTask) -> None:
163161
"""
164162
await self.redis.set(
165163
f"{self.prefix}:{schedule.schedule_id}",
166-
self.serializer.dumpb(model_dump(schedule)),
164+
self.serializer.dumpb(schedule.model_dump(mode="json")),
167165
)
168166

169167
async def get_schedules(self) -> list[ScheduledTask]:
@@ -177,8 +175,7 @@ async def get_schedules(self) -> list[ScheduledTask]:
177175
schedules = []
178176
async for key in self.redis.scan_iter(f"{self.prefix}:*"):
179177
raw_schedule = await self.redis.get(key)
180-
parsed_schedule = model_validate(
181-
ScheduledTask,
178+
parsed_schedule = ScheduledTask.model_validate(
182179
self.serializer.loadb(raw_schedule), # type: ignore[arg-type]
183180
)
184181
schedules.append(parsed_schedule)
@@ -255,7 +252,7 @@ async def add_schedule(self, schedule: ScheduledTask) -> None:
255252
async with self._acquire_master_conn() as redis:
256253
await redis.set(
257254
f"{self.prefix}:{schedule.schedule_id}",
258-
self.serializer.dumpb(model_dump(schedule)),
255+
self.serializer.dumpb(schedule.model_dump(mode="json")),
259256
)
260257

261258
async def get_schedules(self) -> list[ScheduledTask]:
@@ -277,7 +274,7 @@ async def get_schedules(self) -> list[ScheduledTask]:
277274
if buffer:
278275
schedules.extend(await redis.mget(buffer))
279276
return [
280-
model_validate(ScheduledTask, self.serializer.loadb(schedule)) # type: ignore[arg-type]
277+
ScheduledTask.model_validate(self.serializer.loadb(schedule)) # type: ignore[arg-type]
281278
for schedule in schedules
282279
if schedule
283280
]

0 commit comments

Comments
 (0)