Skip to content

Instantly share code, notes, and snippets.

Last active Feb 4, 2022
What would you like to do?
Clean up Python caches and packaging artifacts. See
#!/usr/bin/env python3
## Author: Alex DeLorenzo
## Homepage:
## License: GPLv3
from __future__ import annotations
from typing import Final, AsyncIterable
from shutil import rmtree
from sys import argv
from asyncio import (
gather, to_thread, run, Task,
create_task as to_background
from aiopath import AsyncPath #
from import merge
DELETE_DIRS: Final[set[str]] = {
DEFAULT_DIRS: Final[list[str]] = ['.']
async def delete_path(path: AsyncPath):
if await path.is_dir():
print('Deleting', path)
await to_thread(rmtree, path)
elif await path.is_file():
print('Deleting', path)
await path.unlink(missing_ok=True)
async def get_artifacts(path: AsyncPath) -> AsyncIterable[AsyncPath]:
artifact_gens = (path.glob(pattern) for pattern in DELETE_DIRS)
async for artifact in merge(*artifact_gens):
yield artifact
async def delete_artifacts(path: str):
path = AsyncPath(path)
path_tasks: dict[AsyncPath, Task] = {}
async for path in get_artifacts(path):
if path in path_tasks:
coro = delete_path(path)
path_tasks[path]: Task = to_background(coro)
tasks = path_tasks.values()
await gather(*tasks)
async def main():
paths: list[str]
_, *paths = argv
paths = paths or DEFAULT_DIRS
coros = map(delete_artifacts, paths)
await gather(*coros)
if __name__ == '__main__':
coro = main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment