Skip to content

Instantly share code, notes, and snippets.

@alexdelorenzo
Last active Feb 4, 2022
Embed
What would you like to do?
Clean up Python caches and packaging artifacts. See https://alexdelorenzo.dev/notes/pyclean
#!/usr/bin/env python3
## Author: Alex DeLorenzo
## Homepage: alexdelorenzo.dev
## 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 # https://github.com/alexdelorenzo/aiopath
from aiostream.stream import merge
DELETE_DIRS: Final[set[str]] = {
'build',
'dist',
'*.egg-info',
'**/__pycache__',
}
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:
continue
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()
run(coro)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment