(path: str, record: Dict[str, Any])
| 61 | |
| 62 | |
| 63 | async def append_stats(path: str, record: Dict[str, Any]) -> None: |
| 64 | if not path: |
| 65 | return |
| 66 | lock = _get_lock(path) |
| 67 | async with lock: |
| 68 | os.makedirs(os.path.dirname(path) or '.', exist_ok=True) |
| 69 | data: list = [] |
| 70 | if os.path.exists(path): |
| 71 | try: |
| 72 | with open(path, 'r', encoding='utf-8') as f: |
| 73 | data = json.load(f) or [] |
| 74 | except Exception as exc: |
| 75 | logger.warning( |
| 76 | f'Failed to read stats file {path}, resetting: {exc}') |
| 77 | data = [] |
| 78 | if not isinstance(data, list): |
| 79 | data = [] |
| 80 | |
| 81 | record.setdefault('created_at', datetime.utcnow().isoformat()) |
| 82 | data.append(record) |
| 83 | with open(path, 'w', encoding='utf-8') as f: |
| 84 | json.dump(data, f, ensure_ascii=True, indent=2) |
| 85 | |
| 86 | |
| 87 | def build_timing_record( |
no test coverage detected