Add Parameter pattern to JobQueue.jobs()

telegram/ext/_jobqueue.py

@@ -19,6 +19,7 @@
 """This module contains the classes JobQueue and Job."""
 import asyncio
 import datetime as dtm
+import re
 import weakref
 from typing import TYPE_CHECKING, Any, Generic, Optional, Union, cast, overload
 
@@ -38,6 +39,8 @@ from telegram.ext._extbot import ExtBot
 from telegram.ext._utils.types import CCT, JobCallback
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     if APS_AVAILABLE:
         from apscheduler.job import Job as APSJob
 
@@ -693,22 +696,43 @@ class JobQueue(Generic[CCT]):
             # so give it a tiny bit of time to actually shut down.
             await asyncio.sleep(0.01)
 
-    def jobs(self) -> tuple["Job[CCT]", ...]:
+    def jobs(self, pattern: Union[str, re.Pattern, None] = None) -> tuple["Job[CCT]", ...]:
         """Returns a tuple of all *scheduled* jobs that are currently in the :class:`JobQueue`.
 
+        Args:
+            pattern (:obj:`str` | :obj:`re.Pattern`, optional): A regular expression pattern. If
+                passend, only jobs whose name matches the pattern will be returned.
+                Defaults to :obj:`None`.
+
+                Hint:
+                    This uses :func:`re.search` and not :func:`re.match`.
+
+                .. versionadded:: NEXT.VERSION
+
         Returns:
             tuple[:class:`Job`]: Tuple of all *scheduled* jobs.
         """
-        return tuple(Job.from_aps_job(job) for job in self.scheduler.get_jobs())
+        jobs_generator: Iterable[Job] = (
+            Job.from_aps_job(job) for job in self.scheduler.get_jobs()
+        )
+        if pattern is None:
+            return tuple(jobs_generator)
+        return tuple(
+            job for job in jobs_generator if (job.name and re.compile(pattern).search(job.name))
+        )
 
     def get_jobs_by_name(self, name: str) -> tuple["Job[CCT]", ...]:
-        """Returns a tuple of all *pending/scheduled* jobs with the given name that are currently
+        """Returns a tuple of all *scheduled* jobs with the given name that are currently
         in the :class:`JobQueue`.
 
+        Hint:
+            This method is a convenience wrapper for :meth:`jobs` with a pattern that matches the
+            given name.
+
         Returns:
-            tuple[:class:`Job`]: Tuple of all *pending* or *scheduled* jobs matching the name.
+            tuple[:class:`Job`]: Tuple of all *scheduled* jobs matching the name.
         """
-        return tuple(job for job in self.jobs() if job.name == name)
+        return self.jobs(f"^{re.escape(name)}$")
 
 
 class Job(Generic[CCT]):

tests/ext/test_jobqueue.py

@@ -21,6 +21,7 @@ import calendar
 import datetime as dtm
 import logging
 import platform
+import re
 import time
 
 import pytest
@@ -443,19 +444,34 @@ class TestJobQueue:
 
         await jq.stop()
 
-    async def test_get_jobs(self, job_queue):
+    @pytest.mark.parametrize(
+        "pattern", [None, "not", re.compile("not")], ids=["None", "str", "re.Pattern"]
+    )
+    async def test_get_jobs(self, job_queue, pattern):
         callback = self.job_run_once
 
-        job1 = job_queue.run_once(callback, 10, name="name1")
+        job1 = job_queue.run_once(callback, 10, name="is|a|match")
         await asyncio.sleep(0.03)  # To stablize tests on windows
-        job2 = job_queue.run_once(callback, 10, name="name1")
+        job2 = job_queue.run_once(callback, 10, name="is|a|match")
         await asyncio.sleep(0.03)
-        job3 = job_queue.run_once(callback, 10, name="name2")
+        job3 = job_queue.run_once(callback, 10, name="not|is|a|match")
+        await asyncio.sleep(0.03)
+        job4 = job_queue.run_once(callback, 10, name="is|a|match|not")
+        await asyncio.sleep(0.03)
+        job5 = job_queue.run_once(callback, 10, name="something_else")
+        await asyncio.sleep(0.03)
+        # name-less job
+        job6 = job_queue.run_once(callback, 10)
         await asyncio.sleep(0.03)
 
-        assert job_queue.jobs() == (job1, job2, job3)
-        assert job_queue.get_jobs_by_name("name1") == (job1, job2)
-        assert job_queue.get_jobs_by_name("name2") == (job3,)
+        if pattern is None:
+            assert job_queue.jobs(pattern) == (job1, job2, job3, job4, job5, job6)
+        else:
+            assert job_queue.jobs(pattern) == (job3, job4)
+
+        assert job_queue.jobs() == (job1, job2, job3, job4, job5, job6)
+        assert job_queue.get_jobs_by_name("is|a|match") == (job1, job2)
+        assert job_queue.get_jobs_by_name("something_else") == (job5,)
 
     async def test_job_run(self, app):
         job = app.job_queue.run_repeating(self.job_run_once, 0.02)