Introducing Callback via new typing module

- Explicit callback definition
- simplified delete_callback to be a dumb callback (not chaining)

Author: rgaudin

src/zimscraperlib/filesystem.py

@@ -9,8 +9,6 @@
 
 import os
 import pathlib
-from collections.abc import Callable
-from typing import Any
 
 import magic
 
@@ -44,15 +42,7 @@ def get_content_mimetype(content: bytes | str) -> str:
     return MIME_OVERRIDES.get(detected_mime, detected_mime)
 
 
-def delete_callback(
-    fpath: str | pathlib.Path,
-    callback: Callable | None = None,
-    *callback_args: Any,
-):
-    """helper deleting passed filepath, optionnaly calling an additional callback"""
+def delete_callback(fpath: str | pathlib.Path):
+    """helper deleting passed filepath"""
 
     os.unlink(fpath)
-
-    # call the callback if requested
-    if callback and callable(callback):
-        callback.__call__(*callback_args)

src/zimscraperlib/typing.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Any, NamedTuple
+
+
+class Callback(NamedTuple):
+    func: Callable
+    args: tuple[Any, ...] | None = None
+    kwargs: dict[str, Any] | None = None
+
+    @property
+    def callable(self) -> bool:
+        return callable(self.func)
+
+    def get_args(self) -> tuple[Any, ...]:
+        return self.args or ()
+
+    def get_kwargs(self) -> dict[str, Any]:
+        return self.kwargs or {}
+
+    def call_with(self, *args, **kwargs):
+        self.func.__call__(*args, **kwargs)
+
+    def call(self):
+        self.call_with(*self.get_args(), **self.get_kwargs())

src/zimscraperlib/zim/creator.py

@@ -40,6 +40,7 @@
 )
 from zimscraperlib.i18n import is_valid_iso_639_3
 from zimscraperlib.types import get_mime_for_name
+from zimscraperlib.typing import Callback
 from zimscraperlib.zim.indexing import IndexData
 from zimscraperlib.zim.items import StaticItem
 from zimscraperlib.zim.metadata import (
@@ -341,7 +342,7 @@ def add_item_for(
         should_compress: bool | None = None,
         delete_fpath: bool | None = False,
         duplicate_ok: bool | None = None,
-        callback: Callable | tuple[Callable, Any] | None = None,
+        callbacks: list[Callback] | Callback | None = None,
         index_data: IndexData | None = None,
         auto_index: bool = True,
     ):
@@ -365,6 +366,11 @@ def add_item_for(
         if fpath is None and content is None:
             raise ValueError("One of fpath or content is required")
 
+        if isinstance(callbacks, Callback):
+            callbacks = [callbacks]
+        elif callbacks is None:
+            callbacks = []
+
         mimetype = mimetype_for(
             path=path, content=content, fpath=fpath, mimetype=mimetype
         )
@@ -377,12 +383,7 @@ def add_item_for(
             hints[libzim.writer.Hint.COMPRESS] = should_compress
 
         if delete_fpath and fpath:
-            cb = [delete_callback, fpath]
-            if callback and callable(callback):
-                cb.append(callback)
-            elif callback:
-                cb += list(callback)
-            callback = tuple(cb)
+            callbacks.append(Callback(func=delete_callback, args=(fpath,)))
 
         self.add_item(
             StaticItem(
@@ -395,7 +396,7 @@ def add_item_for(
                 index_data=index_data,
                 auto_index=auto_index,
             ),
-            callback=callback,
+            callbacks=callbacks,
             duplicate_ok=duplicate_ok,
         )
         return path
@@ -404,18 +405,23 @@ def add_item(
         self,
         item: libzim.writer.Item,
         duplicate_ok: bool | None = None,
-        callback: Callable | tuple[Callable, Any] | None = None,
+        callbacks: list[Callback] | Callback | None = None,
     ):
         """Add a libzim.writer.Item
 
         callback: either a single callable or a tuple containing the callable
         as first element then the arguments to pass to the callable.
         Note: you must __not__ include the item itself in those arguments."""
-        if callback:
-            if callable(callback):
-                weakref.finalize(item, callback)
-            else:
-                weakref.finalize(item, *callback)
+        if isinstance(callbacks, Callback):
+            callbacks = [callbacks]
+        elif callbacks is None:
+            callbacks = []
+
+        for callback in callbacks:
+            if callback.callable:
+                weakref.finalize(
+                    item, callback.func, *callback.get_args(), **callback.get_kwargs()
+                )
 
         duplicate_ok = duplicate_ok or self.ignore_duplicates
         try:

tests/filesystem/test_filesystem.py

@@ -43,25 +43,7 @@ def test_mime_overrides(svg_image):
             assert get_content_mimetype(fh.read(64)) == expected_mime
 
 
-def test_delete_callback_with_cb(tmp_path):
-    class Store:
-        called = 0
-
-    def cb(*args):  # noqa: ARG001
-        Store.called += 1
-
-    fpath = tmp_path.joinpath("my-file")
-    with open(fpath, "w") as fh:
-        fh.write("content")
-
-    delete_callback(fpath, cb, fpath.name)
-
-    assert not fpath.exists()
-    assert Store.called
-    assert Store.called == 1
-
-
-def test_delete_callback_without_cb(tmp_path):
+def test_delete_callback(tmp_path):
     fpath = tmp_path.joinpath("my-file")
     with open(fpath, "w") as fh:
         fh.write("content")