Merge pull request #78 from nsteinmetz/patch-1

Graeme22 · web-flow · commit 4436e17eadff · 2025-08-28T16:20:29.000-04:00
Add function name to logs
diff --git a/README.md b/README.md
@@ -79,13 +79,13 @@ Let's see what the output looks like:
 
 ```
 [INFO] 02:14:30: starting worker 3265311d for 2 functions
-[INFO] 02:14:35: task cf0c55387a214320bd23e8987283a562 → worker 3265311d
-[INFO] 02:14:38: task cf0c55387a214320bd23e8987283a562 ← 3
-[INFO] 02:14:40: task 1de3f192ee4a40d4884ebf303874681c → worker 3265311d
-[INFO] 02:14:41: task 1de3f192ee4a40d4884ebf303874681c ← 1
-[INFO] 02:15:00: task 2a4b864e5ecd4fc99979a92f5db3a6e0 → worker 3265311d
+[INFO] 02:14:35: task fetch □ cf0c55387a214320bd23e8987283a562 → worker 3265311d
+[INFO] 02:14:38: task fetch ■ cf0c55387a214320bd23e8987283a562 ← 3
+[INFO] 02:14:40: task fetch □ 1de3f192ee4a40d4884ebf303874681c → worker 3265311d
+[INFO] 02:14:41: task fetch ■ 1de3f192ee4a40d4884ebf303874681c ← 1
+[INFO] 02:15:00: task cronjob □ 2a4b864e5ecd4fc99979a92f5db3a6e0 → worker 3265311d
 Nobody respects the spammish repetition!
-[INFO] 02:15:00: task 2a4b864e5ecd4fc99979a92f5db3a6e0 ← None
+[INFO] 02:15:00: task cronjob ■ 2a4b864e5ecd4fc99979a92f5db3a6e0 ← None
 ```
 ```python
 TaskInfo(fn_name='sleeper', enqueue_time=1751508876961, tries=0, scheduled=datetime.datetime(2025, 7, 3, 2, 14, 39, 961000, tzinfo=datetime.timezone.utc), dependencies=set(), dependents=set())
diff --git a/docs/getting-started.rst b/docs/getting-started.rst
@@ -26,7 +26,7 @@ To start, you'll need to create a ``Worker`` object. At worker creation, you can
        You can also do any startup/shutdown work here
        """
        async with AsyncClient() as http_client:
-           yield Context(http_client)
+           yield WorkerContext(http_client)
 
    worker = Worker(redis_url="redis://localhost:6379", lifespan=lifespan)
 
@@ -66,13 +66,12 @@ Let's see what the output looks like:
 
 .. code-block::
 
-   [INFO] 13:18:07: starting worker ca5bd9eb for 2 functions
-   [INFO] 13:18:10: task 1ab9543aae374bd89713ca00f5c566f9 → worker ca5bd9eb
-   [INFO] 13:18:10: task 1ab9543aae374bd89713ca00f5c566f9 ← 15
-   [INFO] 13:18:14: task cac277a9e3034704a36e67099c1d6f07 → worker ca5bd9eb
-   [INFO] 13:18:15: task cac277a9e3034704a36e67099c1d6f07 ← 303557
+   [INFO] 07:19:48: task fetch □ 45d7ff032e6d42239e9f479a2fc4b70e → worker 12195ce1
+   [INFO] 07:19:48: task fetch ■ 45d7ff032e6d42239e9f479a2fc4b70e ← 15
+   [INFO] 07:19:51: task fetch □ 65e687f9ba644a1fbe23096fa246dfe1 → worker 12195ce1
+   [INFO] 07:19:52: task fetch ■ 65e687f9ba644a1fbe23096fa246dfe1 ← 303659
 
 .. code-block:: python
 
-   TaskInfo(fn_name='fetch', enqueue_time=1751635090933, tries=0, scheduled=datetime.datetime(2025, 7, 4, 13, 18, 13, 933000, tzinfo=datetime.timezone.utc), dependencies=set(), dependents=set())
-   TaskResult(fn_name='fetch', enqueue_time=1751635090933, success=True, result=303557, start_time=1751635094068, finish_time=1751635095130, tries=1, worker_id='ca5bd9eb')
+   TaskInfo(fn_name='fetch', enqueue_time=1756365588232, tries=0, scheduled=datetime.datetime(2025, 8, 28, 7, 19, 51, 232000, tzinfo=datetime.timezone.utc), dependencies=set(), dependents=set())
+   TaskResult(fn_name='fetch', enqueue_time=1756365588232, success=True, result=303659, start_time=1756365591327, finish_time=1756365592081, tries=1, worker_id='12195ce1')
diff --git a/docs/worker.rst b/docs/worker.rst
@@ -36,7 +36,7 @@ Next, create an async context manager to run at worker creation/teardown. Use th
        # here we run code if desired after worker start up
        # yield our dependencies as an instance of the class
        async with AsyncClient() as http_client:
-           yield Context(http_client)
+           yield WorkerContext(http_client)
        # here we run code if desired before worker shutdown
 
 Now, tasks created for the worker will have access to the dependencies like so:
diff --git a/pyproject.toml b/pyproject.toml
@@ -54,10 +54,10 @@ authors = [
     { name = "Graeme Holliday", email = "graeme@tastyware.dev" }
 ]
 dependencies = [
-    "anyio>=4.8.0",
+    "anyio>=4.10.0",
     "coredis>=5.0.1",
     "crontab>=1.0.5",
-    "typer>=0.15.2",
+    "typer>=0.16.1",
     "uvloop>=0.21.0; sys_platform != 'win32'",
     "watchfiles>=1.1.0",
 ]
@@ -71,10 +71,10 @@ benchmark = [
 ]
 web = [
     "async-lru>=2.0.5",
-    "fastapi>=0.115.12",
+    "fastapi>=0.116.1",
     "jinja2>=3.1.6",
     "python-multipart>=0.0.20",
-    "uvicorn>=0.34.2",
+    "uvicorn>=0.35.0",
 ]
 
 [project.scripts]
@@ -89,16 +89,16 @@ Changelog = "https://github.com/tastyware/streaq/releases"
 
 [dependency-groups]
 dev = [
-    "enum-tools[sphinx]>=0.12.0",
+    "enum-tools[sphinx]>=0.13.0",
     "httpx>=0.28.1",
-    "pyright>=1.1.392.post0",
-    "pytest>=8.3.4",
-    "pytest-cov>=6.0.0",
+    "pyright>=1.1.404",
+    "pytest>=8.4.1",
+    "pytest-cov>=6.2.1",
     "pytest-xdist>=3.8.0",
-    "ruff>=0.9.3",
+    "ruff>=0.12.10",
     "sphinx>=8.1.3",
-    "sphinx-immaterial>=0.13.5",
-    "testcontainers[redis]>=4.9.1",
+    "sphinx-immaterial>=0.13.6",
+    "testcontainers[redis]>=4.12.0",
 ]
 
 [tool.pytest.ini_options]
diff --git a/streaq/__init__.py b/streaq/__init__.py
@@ -2,7 +2,7 @@
 
 import coredis
 
-VERSION = "5.2.1"
+VERSION = "5.2.2"
 __version__ = VERSION
 
 logger = logging.getLogger(__name__)
diff --git a/streaq/worker.py b/streaq/worker.py
@@ -405,11 +405,10 @@ def wrapped(fn: AsyncCron[R] | SyncCron[R]) -> RegisteredCron[C, R]:
             )
             if task.fn_name in self.registry:
                 raise StreaqError(
-                    f"A task named {task.fn_name!r} has already been registered!"
+                    f"A task named {task.fn_name} has already been registered!"
                 )
             self.cron_jobs[task.fn_name] = task
             self.registry[task.fn_name] = task
-            logger.debug(f"cron job {task.fn_name} registered in worker {self.id}")
             return task
 
         return wrapped  # type: ignore
@@ -460,10 +459,9 @@ def wrapped(
             )
             if task.fn_name in self.registry:
                 raise StreaqError(
-                    f"A task named {task.fn_name!r} has already been registered!"
+                    f"A task named {task.fn_name} has already been registered!"
                 )
             self.registry[task.fn_name] = task
-            logger.debug(f"task {task.fn_name} registered in worker {self.id}")
             return task
 
         return wrapped  # type: ignore
@@ -637,7 +635,9 @@ async def abort_tasks(self, tasks: set[str]) -> None:
                 and not self._cancel_scopes[task_id].cancel_called
             ):
                 self._cancel_scopes[task_id].cancel()
-                logger.debug(f"aborting task {task_id} in worker {self.id}")
+                logger.debug(
+                    f"task ⊘ {task_id} marked for abortion in worker {self.id}"
+                )
 
     async def schedule_cron_jobs(self) -> None:
         """
@@ -767,7 +767,7 @@ def key(mid: str) -> str:
                 if len(output) > truncate_length:
                     output = f"{output[:truncate_length]}…"
                 if not silent:
-                    logger.info(f"task {task_id} ← {output}")
+                    logger.info(f"task {fn_name} ■ {task_id} ← {output}")
                 if triggers:
                     args = self.serialize(to_tuple(return_value))
                     pipe.set(key(REDIS_PREVIOUS), args, ex=timedelta(minutes=5))
@@ -841,26 +841,26 @@ def key(mid: str) -> str:
         await pipe.execute()
         raw, task_try, abort, active = await asyncio.gather(*commands)
         if not raw:
-            logger.warning(f"task {task_id} expired †")
+            logger.warning(f"task † {task_id} expired")
             return await self.finish_failed_task(
                 msg, StreaqError("Task expired!"), task_try
             )
         if not active:
-            logger.warning(f"task {task_id} reclaimed ↩ from worker {self.id}")
+            logger.warning(f"task ↩ {task_id} reclaimed from worker {self.id}")
             self.counters["relinquished"] += 1
             return None
 
         try:
             data = self.deserialize(raw)
         except StreaqError as e:
-            logger.exception(f"Failed to deserialize task {task_id}!")
+            logger.error(f"task ☒ {task_id} failed to deserialize")
             return await self.finish_failed_task(msg, e, task_try)
 
         if (fn_name := data["f"]) not in self.registry:
-            logger.error(f"Missing function {fn_name}, can't execute task {task_id}!")
+            logger.error(f"task {fn_name} ⊘ {task_id} aborted, missing function")
             return await self.finish_failed_task(
                 msg,
-                StreaqError("Nonexistent function!"),
+                StreaqError(f"Missing function {fn_name}!"),
                 task_try,
                 enqueue_time=data["t"],
                 fn_name=data["f"],
@@ -869,27 +869,27 @@ def key(mid: str) -> str:
 
         if abort:
             if not task.silent:
-                logger.info(f"task {task_id} aborted ⊘ prior to run")
+                logger.info(f"task {fn_name} ⊘ {task_id} aborted prior to run")
             return await self.finish_failed_task(
                 msg,
                 asyncio.CancelledError("Task aborted prior to run!"),
                 task_try,
                 enqueue_time=data["t"],
-                fn_name=data["f"],
+                fn_name=fn_name,
                 silent=task.silent,
                 ttl=task.ttl,
             )
         if task.max_tries and task_try > task.max_tries:
             if not task.silent:
                 logger.warning(
-                    f"task {task_id} failed × after {task.max_tries} retries"
+                    f"task {fn_name} × {task_id} failed after {task.max_tries} retries"
                 )
             return await self.finish_failed_task(
                 msg,
-                StreaqError(f"Max retry attempts reached for task {task_id}!"),
+                StreaqError("Max retry attempts reached for task!"),
                 task_try,
                 enqueue_time=data["t"],
-                fn_name=data["f"],
+                fn_name=fn_name,
                 silent=task.silent,
                 ttl=task.ttl,
             )
@@ -901,7 +901,7 @@ def key(mid: str) -> str:
         after = data.get("A")
         pipe = await self.redis.pipeline(transaction=True)
         if task.unique:
-            lock_key = self.prefix + REDIS_UNIQUE + task.fn_name
+            lock_key = self.prefix + REDIS_UNIQUE + fn_name
             locked = pipe.set(
                 lock_key, task_id, get=True, condition=PureToken.NX, pxat=timeout
             )
@@ -917,7 +917,8 @@ def key(mid: str) -> str:
             if existing and existing != task_id:
                 if not task.silent:
                     logger.warning(
-                        f"unique task {task_id} clashed ↯ with running task {existing}"
+                        f"task {fn_name} ↯ {task_id} clashed with unique task "
+                        f"{existing}"
                     )
                 return await self.finish_failed_task(
                     msg,
@@ -927,7 +928,7 @@ def key(mid: str) -> str:
                     ),
                     task_try,
                     enqueue_time=data["t"],
-                    fn_name=data["f"],
+                    fn_name=fn_name,
                     silent=task.silent,
                     ttl=task.ttl,
                 )
@@ -959,7 +960,7 @@ async def _fn(*args: Any, **kwargs: Any) -> Any:
                 return await task.fn(*args, **kwargs)
 
         if not task.silent:
-            logger.info(f"task {task_id} → worker {self.id}")
+            logger.info(f"task {task.fn_name} □ {task_id} → worker {self.id}")
 
         wrapped = _fn
         for middleware in reversed(self.middlewares):
@@ -978,7 +979,7 @@ async def _fn(*args: Any, **kwargs: Any) -> Any:
                 success = False
                 done = True
                 if not task.silent:
-                    logger.info(f"task {task_id} aborted ⊘")
+                    logger.info(f"task {task.fn_name} ⊘ {task_id} aborted")
                     self.counters["aborted"] += 1
                     self.counters["failed"] -= 1  # this will get incremented later
         except StreaqRetry as e:
@@ -988,22 +989,26 @@ async def _fn(*args: Any, **kwargs: Any) -> Any:
                 schedule = datetime_ms(e.schedule)
                 if not task.silent:
                     logger.exception(f"Retrying task {task_id}!")
-                    logger.info(f"retrying ↻ task {task_id} at {schedule}")
+                    logger.info(
+                        f"task {task.fn_name} ↻ {task_id} retrying at {schedule}"
+                    )
             else:
                 delay = to_ms(e.delay) if e.delay is not None else task_try**2 * 1000
                 schedule = now_ms() + delay
                 if not task.silent:
                     logger.exception(f"Retrying task {task_id}!")
-                    logger.info(f"retrying ↻ task {task_id} in {delay}s")
+                    logger.info(f"task {task.fn_name} ↻ {task_id} retrying in {delay}s")
         except TimeoutError as e:
             if not task.silent:
-                logger.error(f"task {task_id} timed out …")
+                logger.error(f"task {task.fn_name} … {task_id} timed out")
             result = e
             success = False
             done = True
         except asyncio.CancelledError:
             if not task.silent:
-                logger.info(f"task {task_id} cancelled, will be retried ↻")
+                logger.info(
+                    f"task {task.fn_name} ↻ {task_id} cancelled, will be retried"
+                )
             success = False
             done = False
             raise  # best practice from anyio docs
@@ -1012,8 +1017,8 @@ async def _fn(*args: Any, **kwargs: Any) -> Any:
             success = False
             done = True
             if not task.silent:
-                logger.info(f"task {task_id} failed ×")
                 logger.exception(f"Task {task_id} failed!")
+                logger.info(f"task {task.fn_name} × {task_id} failed")
         finally:
             with CancelScope(shield=True):
                 finish_time = now_ms()
@@ -1057,7 +1062,7 @@ async def fail_task_dependents(self, dependents: list[str]) -> None:
         self.counters["failed"] += len(dependents)
         to_delete: list[KeyT] = []
         for dep_id in dependents:
-            logger.info(f"task {dep_id} dependency failed ×")
+            logger.info(f"task dependent × {dep_id} failed")
             to_delete.append(self.prefix + REDIS_TASK + dep_id)
             pipe.set(self.results_key + dep_id, result, ex=300)
             pipe.publish(self._channel_key + dep_id, result)
diff --git a/uv.lock b/uv.lock
