helpers.py (1007B)
1 from .io import open 2 from .once import once 3 from .rebuild import cache_conditionally, rerun_if_changed, rerun_always 4 import hashlib 5 import os 6 7 8 async def file_modtime(f: int | str | bytes | os.PathLike[str] | os.PathLike[bytes]): 9 return os.stat(f).st_mtime_ns 10 11 12 @once() 13 @cache_conditionally(lambda f, *args: (f.name, *args)) 14 async def _file_hash(f: open, skip_if_modtime_matches=True): 15 if skip_if_modtime_matches: 16 rerun_if_changed()(await file_modtime(f.fileno())) 17 else: 18 rerun_always() 19 h = hashlib.sha256() 20 while True: 21 chunk = f.read1() 22 if chunk: 23 h.update(chunk) 24 else: 25 break 26 d = h.hexdigest() 27 # print("hash", f.name, d) 28 return d 29 30 31 async def file_hash(f: open | bytes | str, skip_if_modtime_matches=True): 32 if isinstance(f, bytes) or isinstance(f, str): 33 with open(f, "rb") as _f: 34 return await _file_hash(_f, skip_if_modtime_matches) 35 return await _file_hash(f, skip_if_modtime_matches)