Buckets:
| import fcntl | |
| import json | |
| import os | |
| import time | |
| from datetime import UTC, datetime | |
| from pathlib import Path | |
| START = time.monotonic() | |
| def record(kind: str, **fields: object) -> dict[str, object]: | |
| event: dict[str, object] = { | |
| "timestamp": datetime.now(UTC).isoformat(), | |
| "process_elapsed_seconds": time.monotonic() - START, | |
| "kind": kind, | |
| **fields, | |
| } | |
| path = Path(os.getenv("AUTOJEV_EVENTS", "progress/events.jsonl")) | |
| path.parent.mkdir(parents=True, exist_ok=True) | |
| with path.open("a") as stream: | |
| fcntl.flock(stream.fileno(), fcntl.LOCK_EX) | |
| stream.write(json.dumps(event, ensure_ascii=False) + "\n") | |
| stream.flush() | |
| fcntl.flock(stream.fileno(), fcntl.LOCK_UN) | |
| return event | |
Xet Storage Details
- Size:
- 756 Bytes
- Xet hash:
- 091a2fbddc69e99c2a947fa14548dba22d933601e3e9d5ea1004197bba43d73c
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.