Zipping Many Files

Question

My question:

I'm wondering if anyone more experienced with using asyncio would consider the way I wrote the code to be reasonably canonical asyncio code. Perhaps I'm using overcomplicated or unnecessarily low-level mechanisms, or perhaps there are far more readable or safer ways to achieve the same result.

Nota Bene: The code below creates some large-ish test files for the duration of the run.

Python Version: 3.12

What the code does:

• I need to zip up a bunch of directories (dirs).
• The code below creates some test dirs, and puts files of varying sizes inside, then zips them all up.
• The real-life code gets lists of dirs from multiple sources, otherwise the remaining code is more or less identical.
• I used a TaskGroup because its seemed the easiest structure to just fire off all the dirs from my lists of dirs asynchronously.

Code Landmarks:

• create_zip function. The original synchronous function that zips a dir the way I want it.
• async_create_zip an async wrapper for create_zip.
• make_packages - The non-test-setup code starts at the TaskGroup context manager.

My background:

• I have not used python's asyncio module before.
• I have managed to write the code to zip my stuff and make it work using asyncio.

import asyncio
import shutil
import tempfile
from datetime import datetime
from pathlib import Path


def create_zip(dir_to_zip: Path, output_dir: Path):
    """Zip a directory.

    Args:
        dir_to_zip (Path): Directory to zip.
        output_dir (Path): Dir where the zip will be written.
    """
    try:
        dest_file_no_ext = output_dir.joinpath(dir_to_zip.name)
        print(
            f"{datetime.now()}: Creating ZIP archive: {dest_file_no_ext.name}.zip from "
            f"{dir_to_zip.name}"
        )

        # You could swap the call below with
        # sleep(10) or something to test without actually creating files.

        shutil.make_archive(
            dest_file_no_ext.as_posix(),  # Name of zip without ext.
            "zip",
            dir_to_zip.parent,  # Dir to zip from.
            dir_to_zip.name,  # Dir to include in the zip
        )
        print(f"{datetime.now()}: Created: {dest_file_no_ext.name}.zip.")
    except Exception as e:
        print(
            "asyncio.TaskGroup fails in its entirety if any exceptions occur "
            "but I would simply swallow and log any exceptions here."
        )


async def async_create_zip(dir_to_zip: Path, output_dir: Path):
    loop = asyncio.get_event_loop()
    create_zip_task = loop.run_in_executor(
        None, lambda: create_zip(dir_to_zip, output_dir)
    )
    await create_zip_task


async def make_packages():
    def create_fake_dir(fake_dir: Path, num_bytes: int) -> Path:
        # Test function to create directories containing
        # files of various sizes.
        fake_dir.mkdir()

        with fake_dir.joinpath("fakefile.bin").open("wb") as outstream:
            outstream.write(b"\x00" * num_bytes)

        return fake_dir

    with tempfile.TemporaryDirectory() as td:
        tempdir = Path(td)

        input_dir = tempdir.joinpath("input")
        output_dir = tempdir.joinpath("output")

        # In my real code, the dirs come from multiple locations.
        test_dirs = [
            create_fake_dir(tempdir.joinpath("dir1"), 100000000),  # 100mb
            create_fake_dir(tempdir.joinpath("dir2"), 1000000),  # 1mb
            create_fake_dir(tempdir.joinpath("dir3"), 100000000),  # 100mb
            create_fake_dir(tempdir.joinpath("dir4"), 500000),  # 0.5mb
        ]

        # Test synchronous.
        print("Synchronous Test:")
        start = datetime.now()
        for test_dir in test_dirs:
            create_zip(test_dir, output_dir)
        end = datetime.now()
        total = (end - start).total_seconds()
        print(f"total time: {total}")

        print("Asynchronous Test:")
        start = datetime.now()
        async with asyncio.TaskGroup() as tg:
            # Fire off create_zip tasks for every package file we find.
            # These tasks will be run asynchronously - so as to not pointlessly
            # wait for large zip files to be packed before packing others.

            for test_dir in test_dirs:
                tg.create_task(async_create_zip(test_dir, output_dir))
        end = datetime.now()
        total = (end - start).total_seconds()
        print(f"total time: {total}")

def run():
    # Creating zips might be a bit slow, so lets try
    # and run zip creation asynchronously.
    asyncio.run(make_packages())


if __name__ == "__main__":
    run()

Output:

Synchronous Test:
2024-08-15 11:51:54.214308: Creating ZIP archive: dir1.zip from dir1
2024-08-15 11:51:54.732268: Created: dir1.zip.
2024-08-15 11:51:54.732391: Creating ZIP archive: dir2.zip from dir2
2024-08-15 11:51:54.745326: Created: dir2.zip.
2024-08-15 11:51:54.745402: Creating ZIP archive: dir3.zip from dir3
2024-08-15 11:51:55.292587: Created: dir3.zip.
2024-08-15 11:51:55.292659: Creating ZIP archive: dir4.zip from dir4
2024-08-15 11:51:55.303336: Created: dir4.zip.
total time: 1.0893
Asynchronous Test:
2024-08-15 11:51:55.313898: Creating ZIP archive: dir1.zip from dir1
2024-08-15 11:51:55.314744: Creating ZIP archive: dir2.zip from dir2
2024-08-15 11:51:55.316534: Creating ZIP archive: dir3.zip from dir3
2024-08-15 11:51:55.317386: Creating ZIP archive: dir4.zip from dir4
2024-08-15 11:51:55.336522: Created: dir4.zip.
2024-08-15 11:51:55.342137: Created: dir2.zip.
2024-08-15 11:51:55.920301: Created: dir3.zip.
2024-08-15 11:51:55.922482: Created: dir1.zip.
total time: 0.619541

Answer 1

swallowing exceptions

    except Exception as e:
        print(
            "asyncio.TaskGroup fails in its entirety if any exceptions occur "
            "but I would simply swallow and log any exceptions here."
        )

Swallowing exceptions is seldom the right answer. Without seeing any further details of your use case, I'd say raise e at the end of this logging handler would be the better choice. Also, as written, except Exception suffices, since we're not using e.

The DbC idea here is we've promised a post-condition, we will have created a zip, made an archive, upon return. If that turns out to be impossible, then blowing up with an exception is the proper way to explain to caller, "whoops, sorry, all bets are off!"

optional type annotations

def create_zip(dir_to_zip: Path, output_dir: Path):

That signature is brilliant, very clear, thank you for using Path.

If we tacked on an annotation of "we're evaluating this for side effects" then "$ mypy --strict *.py" would successfully lint this, putting us in a good position to notice minor errors right after making some new edit.

def create_zip(dir_to_zip: Path, output_dir: Path) -> None:

Similarly for other functions.

BTW in make_packages() there's no need to define input_dir, as it is never used.

slash

Certainly this works:

        ... = output_dir.joinpath(dir_to_zip.name)

But it would be more convenient to use the / slash operator in that expression.

        ... = output_dir / dir_to_zip.name

Also, based on the ..._no_ext identifier, it appears you wanted .stem() rather than .name(). For this particular test data there was no . dot, so it didn't make a difference.

It's unclear why you tacked on the .as_posix() call, as make_archive() is happy to accept a Path parameter. It will turn it into a str on the inside, with either fspath() or abspath().

docstring

    def create_fake_dir( ... ) -> Path:
        # Test function to create directories containing
        # files of various sizes.

Thank you for the very nice # comment. But it should be a """docstring""", please.

---

reasonably canonical asyncio code?

Yes, it looks fine to me. Good job!

Answer 2

In addition to the good advice in the other answer, there area a few more minor points to consider.

Readability

For large numbers like 1000000, it is nice to use the underscore separator: 1_000_000.

        create_fake_dir(tempdir.joinpath("dir1"), 100_000_000),  # 100mb
        create_fake_dir(tempdir.joinpath("dir2"),   1_000_000),  # 1mb
        create_fake_dir(tempdir.joinpath("dir3"), 100_000_000),  # 100mb
        create_fake_dir(tempdir.joinpath("dir4"),     500_000),  # 0.5mb

Comment or feature

If you felt it was important enough to leave the following comment in your code, I suggest removing the comment and adding it as a feature:

    # You could swap the call below with
    # sleep(10) or something to test without actually creating files.

Perhaps pass in a new option to the create_zip function.