(#64) Support tasks retry & propagate raised exception #65
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #65 +/- ##
==========================================
+ Coverage 99.21% 99.59% +0.38%
==========================================
Files 18 21 +3
Lines 635 988 +353
==========================================
+ Hits 630 984 +354
+ Misses 5 4 -1 ☔ View full report in Codecov by Sentry. |
imranariffin
force-pushed
the
64/support-retry-on-tasks
branch
from
aaeb820
to
bff64dc
Compare
imranariffin
changed the title
imranariffin
force-pushed
the
64/support-retry-on-tasks
branch
2 times, most recently
from
d09a7eb
to
8a2e01a
Compare
For documentation, see: 1. Docstring of `task.task` 2. Tests in `tests.test_task` e.g. `test_retry_as_per_task_definition` 3. Sample usages in `tests.apps.simple_app` e.g. `append_to_file` Changelist: * Formalize serialization and deserialization * Serialize & deserialize exceptions correctly * Encapsulate retry & retry_on in a new dict 'options' * Implement serde for AsyncResult * Ensure generated file deleted after test * Add jsonpickle to toml file * Exclude `if TYPE_CHECKING:` from coverage * Add test for singleton * Add logging for worker * Wrap all constants inside `Config` class Signed-off-by: Imran Ariffin <[email protected]>
imranariffin
force-pushed
the
64/support-retry-on-tasks
branch
from
8a2e01a
to
4d4c0c0
Compare
There was a problem hiding this comment.
Should probably rename this to config.py
| return _TASKS_CHANNEL | ||
|
|
||
| @staticmethod | ||
| def results_channel_template() -> str: |
There was a problem hiding this comment.
Maybe better to split this into two files, one for config & one for constants
There was a problem hiding this comment.
Hmm maybe that's a good idea
src/aiotaskq/task.py
Outdated
| self.ready = ready | ||
| self.result = result | ||
| self.error = error | ||
| self.pubsub = PubSub.get(url=Config.broker_url(), poll_interval_s=0.01) |
There was a problem hiding this comment.
Can we remove this since I think it's not used here?
There was a problem hiding this comment.
Makes sense, got lost during refactoring
| self.pubsub = PubSub.get(url=Config.broker_url(), poll_interval_s=0.01) | ||
|
|
||
| @classmethod | ||
| async def from_publisher(cls, task_id: str) -> "AsyncResult": |
There was a problem hiding this comment.
I feel like we should move this polling logic elsewhere & keep AsyncResult as a simple result definition
There was a problem hiding this comment.
I was thinking the same but I couldn't find a way to do it without breaking this kind of interface:
@task()
async def some_task() -> int:
return 42
some_awaitable: Awaitable[int] = some_task.apply_async()
ret: int = await some_awaitable
assert ret == 42
# Or
# ret: int = await some_task.apply_async()
# assert ret == 42Let me think about this more
| await pubsub.publish(channel=result_channel, message=task_result) | ||
| retry = False | ||
| error = None | ||
| retries: int = None |
There was a problem hiding this comment.
Why can't we just initiate this as 0?
There was a problem hiding this comment.
I was thinking of differentiating between "there will be a first retry (0)" vs "there will be no retry at all (None)".
src/aiotaskq/worker.py
Outdated
| # Retry if still within retry limit | ||
| if retry: | ||
| async with redis.from_url(url="redis://localhost:6379") as redis_client: | ||
| retries = int(await redis_client.get(f"retry:{task.id}") or 0) |
There was a problem hiding this comment.
If redis_client.get keeps returning None or something, it might mean we will never stop retrying
(#64) Support tasks retry & propagate raised exception
For documentation, see:
task.task(which will point tointerfaces.TaskOptionsand in turninterfaces.RetryOptions)tests.test_taske.g.test_retry_as_per_task_definitiontests.apps.simple_appe.g.append_to_fileChangelist:
if TYPE_CHECKING:from coverageConfigclass