Skip to content

Commit 59a932a

Browse files
harshil21harshil21
authored
Pass name to asyncio.create_task (#3759)
1 parent 5194025 commit 59a932a

8 files changed

Lines changed: 76 additions & 15 deletions

File tree

‎telegram/ext/_application.py‎

Lines changed: 29 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -568,9 +568,8 @@ async def start(self) -> None:
568568
try:
569569
if self.persistence:
570570
self.__update_persistence_task = asyncio.create_task(
571-
self._persistence_updater()
572-
# TODO: Add this once we drop py3.7
573-
# name=f'Application:{self.bot.id}:persistence_updater'
571+
self._persistence_updater(),
572+
name=f"Application:{self.bot.id}:persistence_updater",
574573
)
575574
_LOGGER.debug("Loop for updating persistence started")
576575

@@ -579,9 +578,7 @@ async def start(self) -> None:
579578
_LOGGER.debug("JobQueue started")
580579

581580
self.__update_fetcher_task = asyncio.create_task(
582-
self._update_fetcher(),
583-
# TODO: Add this once we drop py3.7
584-
# name=f'Application:{self.bot.id}:update_fetcher'
581+
self._update_fetcher(), name=f"Application:{self.bot.id}:update_fetcher"
585582
)
586583
_LOGGER.info("Application started")
587584

@@ -955,6 +952,8 @@ def create_task(
955952
self,
956953
coroutine: Union[Generator[Optional["asyncio.Future[object]"], None, RT], Awaitable[RT]],
957954
update: Optional[object] = None,
955+
*,
956+
name: Optional[str] = None,
958957
) -> "asyncio.Task[RT]":
959958
"""Thin wrapper around :func:`asyncio.create_task` that handles exceptions raised by
960959
the :paramref:`coroutine` with :meth:`process_error`.
@@ -977,16 +976,22 @@ def create_task(
977976
:attr:`chat_data` and :attr:`user_data` entries will be updated in the next run of
978977
:meth:`update_persistence` after the :paramref:`coroutine` is finished.
979978
979+
Keyword Args:
980+
name (:obj:`str`, optional): The name of the task.
981+
982+
.. versionadded:: NEXT.VERSION
983+
980984
Returns:
981985
:class:`asyncio.Task`: The created task.
982986
"""
983-
return self.__create_task(coroutine=coroutine, update=update)
987+
return self.__create_task(coroutine=coroutine, update=update, name=name)
984988

985989
def __create_task(
986990
self,
987991
coroutine: Union[Generator[Optional["asyncio.Future[object]"], None, RT], Awaitable[RT]],
988992
update: Optional[object] = None,
989993
is_error_handler: bool = False,
994+
name: Optional[str] = None,
990995
) -> "asyncio.Task[RT]":
991996
# Unfortunately, we can't know if `coroutine` runs one of the error handler functions
992997
# but by passing `is_error_handler=True` from `process_error`, we can make sure that we
@@ -995,7 +1000,8 @@ def __create_task(
9951000
task: "asyncio.Task[RT]" = asyncio.create_task(
9961001
self.__create_task_callback(
9971002
coroutine=coroutine, update=update, is_error_handler=is_error_handler
998-
)
1003+
),
1004+
name=name,
9991005
)
10001006

10011007
if self.running:
@@ -1076,6 +1082,7 @@ async def _update_fetcher(self) -> None:
10761082
self.create_task(
10771083
self.__process_update_wrapper(update),
10781084
update=update,
1085+
name=f"Application:{self.bot.id}:process_concurrent_update",
10791086
)
10801087
else:
10811088
await self.__process_update_wrapper(update)
@@ -1132,7 +1139,12 @@ async def process_update(self, update: object) -> None:
11321139
and self.bot.defaults
11331140
and not self.bot.defaults.block
11341141
):
1135-
self.create_task(coroutine, update=update)
1142+
self.create_task(
1143+
coroutine,
1144+
update=update,
1145+
name=f"Application:{self.bot.id}:process_update_non_blocking"
1146+
f":{handler}",
1147+
)
11361148
else:
11371149
any_blocking = True
11381150
await coroutine
@@ -1203,7 +1215,10 @@ def add_handler(self, handler: BaseHandler[Any, CCT], group: int = DEFAULT_GROUP
12031215
f"can not be persistent if application has no persistence"
12041216
)
12051217
if self._initialized:
1206-
self.create_task(self._add_ch_to_persistence(handler))
1218+
self.create_task(
1219+
self._add_ch_to_persistence(handler),
1220+
name=f"Application:{self.bot.id}:add_handler:conversation_handler_after_init",
1221+
)
12071222
warn(
12081223
"A persistent `ConversationHandler` was passed to `add_handler`, "
12091224
"after `Application.initialize` was called. This is discouraged."
@@ -1683,7 +1698,10 @@ async def process_error(
16831698
and not self.bot.defaults.block
16841699
):
16851700
self.__create_task(
1686-
callback(update, context), update=update, is_error_handler=True
1701+
callback(update, context),
1702+
update=update,
1703+
is_error_handler=True,
1704+
name=f"Application:{self.bot.id}:process_error:non_blocking",
16871705
)
16881706
else:
16891707
try:

‎telegram/ext/_conversationhandler.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -831,6 +831,7 @@ async def handle_update( # type: ignore[override]
831831
update, application, handler_check_result, context
832832
),
833833
update=update,
834+
name=f"ConversationHandler:{update.update_id}:handle_update:non_blocking_cb",
834835
)
835836
except ApplicationHandlerStop as exception:
836837
new_state = exception.state
@@ -856,6 +857,7 @@ async def handle_update( # type: ignore[override]
856857
new_state, application, update, context, conversation_key
857858
),
858859
update=update,
860+
name=f"ConversationHandler:{update.update_id}:handle_update:timeout_job",
859861
)
860862
else:
861863
self._schedule_job(new_state, application, update, context, conversation_key)

‎telegram/ext/_jobqueue.py‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -797,7 +797,10 @@ async def _run(
797797
await context.refresh_data()
798798
await self.callback(context)
799799
except Exception as exc:
800-
await application.create_task(application.process_error(None, exc, job=self))
800+
await application.create_task(
801+
application.process_error(None, exc, job=self),
802+
name=f"Job:{self.id}:run:process_error",
803+
)
801804
finally:
802805
# This is internal logic of application - let's keep it private for now
803806
application._mark_for_persistence_update(job=self) # pylint: disable=protected-access

‎telegram/ext/_updater.py‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -363,7 +363,8 @@ def default_error_callback(exc: TelegramError) -> None:
363363
on_err_cb=error_callback or default_error_callback,
364364
description="getting Updates",
365365
interval=poll_interval,
366-
)
366+
),
367+
name="Updater:start_polling:polling_task",
367368
)
368369

369370
if ready is not None:

‎tests/ext/test_application.py‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -453,6 +453,8 @@ async def callback(u, c):
453453
async with app:
454454
await app.start()
455455
assert app.running
456+
tasks = asyncio.all_tasks()
457+
assert any(":update_fetcher" in task.get_name() for task in tasks)
456458
if job_queue:
457459
assert app.job_queue.scheduler.running
458460
else:
@@ -551,7 +553,6 @@ async def test_add_remove_handler_non_default_group(self, app):
551553
app.remove_handler(handler)
552554
app.remove_handler(handler, group=2)
553555

554-
#
555556
async def test_handler_order_in_group(self, app):
556557
app.add_handler(MessageHandler(filters.PHOTO, self.callback_set_count(1)))
557558
app.add_handler(MessageHandler(filters.ALL, self.callback_set_count(2)))
@@ -964,6 +965,8 @@ async def callback(update, context):
964965
await app.update_queue.put(1)
965966
task = asyncio.create_task(app.stop())
966967
await asyncio.sleep(0.05)
968+
tasks = asyncio.all_tasks()
969+
assert any(":process_update_non_blocking" in t.get_name() for t in tasks)
967970
assert self.count == 1
968971
# Make sure that app stops only once all non blocking callbacks are done
969972
assert not task.done()
@@ -1029,6 +1032,8 @@ async def normal_error_handler(update, context):
10291032
await app.update_queue.put(self.message_update)
10301033
task = asyncio.create_task(app.stop())
10311034
await asyncio.sleep(0.05)
1035+
tasks = asyncio.all_tasks()
1036+
assert any(":process_error:non_blocking" in t.get_name() for t in tasks)
10321037
assert self.count == 42
10331038
assert self.received is None
10341039
event.set()
@@ -1196,7 +1201,8 @@ async def callback():
11961201
self.count = 42
11971202
return 43
11981203

1199-
task = app.create_task(callback())
1204+
task = app.create_task(callback(), name="test_task")
1205+
assert task.get_name() == "test_task"
12001206
await asyncio.sleep(0.01)
12011207
assert not task.done()
12021208
out = await task
@@ -1377,6 +1383,8 @@ async def callback(u, c):
13771383
assert not events[i].is_set()
13781384

13791385
await asyncio.sleep(0.9)
1386+
tasks = asyncio.all_tasks()
1387+
assert any(":process_concurrent_update" in task.get_name() for task in tasks)
13801388
for i in range(app.update_processor.max_concurrent_updates):
13811389
assert events[i].is_set()
13821390
for i in range(

‎tests/ext/test_basepersistence.py‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -552,6 +552,8 @@ async def test_add_conversation_handler_after_init(self, papp: Application, recw
552552
papp.add_handler(conversation)
553553

554554
assert len(recwarn) >= 1
555+
tasks = asyncio.all_tasks()
556+
assert any("conversation_handler_after_init" in t.get_name() for t in tasks)
555557
found = False
556558
for warning in recwarn:
557559
if "after `Application.initialize` was called" in str(warning.message):
@@ -584,6 +586,28 @@ async def test_add_conversation_handler_without_name(self, papp: Application):
584586
with pytest.raises(ValueError, match="when handler is unnamed"):
585587
papp.add_handler(build_conversation_handler(name=None, persistent=True))
586588

589+
@pytest.mark.parametrize(
590+
"papp",
591+
[
592+
PappInput(update_interval=0.0),
593+
],
594+
indirect=True,
595+
)
596+
async def test_update_persistence_called(self, papp: Application, monkeypatch):
597+
"""Tests if Application.update_persistence is called from app.start()"""
598+
called = asyncio.Event()
599+
600+
async def update_persistence(*args, **kwargs):
601+
called.set()
602+
603+
monkeypatch.setattr(papp, "update_persistence", update_persistence)
604+
async with papp:
605+
await papp.start()
606+
tasks = asyncio.all_tasks()
607+
assert any(":persistence_updater" in task.get_name() for task in tasks)
608+
assert await called.wait()
609+
await papp.stop()
610+
587611
@pytest.mark.flaky(3, 1)
588612
@pytest.mark.parametrize(
589613
"papp",

‎tests/ext/test_conversationhandler.py‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2068,6 +2068,9 @@ async def callback(_, __):
20682068
assert conv_handler.check_update(Update(0, message=message))
20692069
await app.process_update(Update(0, message=message))
20702070
await asyncio.sleep(0.7)
2071+
tasks = asyncio.all_tasks()
2072+
assert any(":handle_update:non_blocking_cb" in t.get_name() for t in tasks)
2073+
assert any(":handle_update:timeout_job" in t.get_name() for t in tasks)
20712074
assert not self.is_timeout
20722075
event.set()
20732076
await asyncio.sleep(0.7)

‎tests/ext/test_updater.py‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -248,6 +248,8 @@ async def delete_webhook(*args, **kwargs):
248248
# We call the same logic twice to make sure that restarting the updater works as well
249249
await updater.start_polling(drop_pending_updates=drop_pending_updates)
250250
assert updater.running
251+
tasks = asyncio.all_tasks()
252+
assert any("Updater:start_polling:polling_task" in t.get_name() for t in tasks)
251253
await updates.join()
252254
await updater.stop()
253255
assert not updater.running

0 commit comments

Comments
 (0)