Coverage for src / quber / files / cache.py: 72%

1"""TTL-aware S3 ingestion cache (QUE-223). 

2 

3Registers a persistent-cache ``cloudpathlib.S3Client`` as the process-global 

4default for ``s3://`` URIs, so any ``CloudPath("s3://...")`` / 

5``AnyPath("s3://...")`` materializes through a project-local cache dir instead 

6of re-downloading every run. Source-change correctness is cloudpathlib's 

7built-in ETag check; disk hygiene is a wall-clock TTL enforced by a single 

8startup sweep. 

9 

10No subclass, no per-access hook — one module, two functions. The cache dir, 

11TTL, and (optional) region come from ``get_settings().s3_cache`` (QUE-228); 

12each function still accepts explicit overrides for tests: 

13 

14- ``evict_stale_s3_cache()`` — startup sweep. Walks the cache dir and unlinks 

15 any file whose access time is older than the TTL. This is the *only* TTL 

16 reclamation path: cloudpathlib's ETag check handles source correctness; the 

17 sweep handles "downloaded once, never touched again" files that would 

18 otherwise live in the cache forever. 

19- ``init_s3_cache()`` — package-import entrypoint. Runs the sweep, then 

20 constructs and registers the default S3 client. Called once from 

21 ``quber/__init__.py``. 

22 

23TTL is keyed off ``st_atime``, so re-processing a cached document resets its 

24window: actively used files stay warm, abandoned ones age out after the TTL. 

25On ``noatime``-mounted volumes atime is frozen and TTL degrades to "time since 

26download" — accepted rather than adding mount-detection complexity. 

27""" 

28 

29from __future__ import annotations 

30 

31import time 

32from pathlib import Path 

33from typing import Sequence 

34 

35from cloudpathlib import AnyPath, S3Client, S3Path 

36from loguru import logger 

37 

38from quber.settings import get_settings 

39 

40 

41def evict_stale_s3_cache( 

42 cache_dir: Path | None = None, 

43 ttl_seconds: float | None = None, 

44) -> int: 

45 """Delete cached files whose atime is older than ``ttl_seconds``. 

46 

47 ``cache_dir`` / ``ttl_seconds`` default to ``get_settings().s3_cache`` when 

48 not passed explicitly. Walks ``cache_dir`` once with ``rglob('*')`` and 

49 unlinks regular files past the TTL. Returns the count evicted. Idempotent 

50 and safe to call when the cache directory does not exist yet (returns 0). 

51 Linear in the cached-file count; sub-second even at thousands. 

52 """ 

53 s3 = get_settings().s3_cache 

54 cache_dir = cache_dir if cache_dir is not None else s3.cache_dir 

55 ttl_seconds = ttl_seconds if ttl_seconds is not None else s3.ttl_seconds 

56 if not cache_dir.exists(): 

57 return 0 

58 cutoff = time.time() - ttl_seconds 

59 evicted = 0 

60 for p in cache_dir.rglob("*"): 

61 if p.is_file() and p.stat().st_atime < cutoff: 

62 p.unlink() 

63 evicted += 1 

64 return evicted 

65 

66 

67def init_s3_cache( 

68 cache_dir: Path | None = None, 

69 ttl_seconds: float | None = None, 

70) -> None: 

71 """Prune the stale cache, then register the default S3 client. 

72 

73 Called once at package import (see ``quber/__init__.py``). ``cache_dir`` / 

74 ``ttl_seconds`` default to ``get_settings().s3_cache``. Two ordered steps: 

75 

76 1. ``evict_stale_s3_cache()`` — bound disk usage for keys not re-accessed 

77 between runs. 

78 2. Construct a persistent-cache ``S3Client`` rooted at ``cache_dir`` and 

79 publish it via ``set_as_default_client()`` so plain 

80 ``CloudPath("s3://...")`` / ``AnyPath("s3://...")`` picks it up. When 

81 ``s3_cache.region`` is set, the client is bound to a region-scoped boto3 

82 session; otherwise region resolves from boto3's own chain. 

83 

84 Constructing the client requires neither AWS credentials nor network access 

85 (boto3 resolves both lazily on first GET), so this is safe to run at import 

86 on any host, including CPU-only / credential-less dev environments. 

87 """ 

88 s3 = get_settings().s3_cache 

89 cache_dir = cache_dir if cache_dir is not None else s3.cache_dir 

90 ttl_seconds = ttl_seconds if ttl_seconds is not None else s3.ttl_seconds 

91 evicted = evict_stale_s3_cache(cache_dir, ttl_seconds) 

92 if evicted: 

93 logger.debug("Evicted {} stale file(s) from S3 cache at {}", evicted, cache_dir) 

94 if s3.region: 

95 import boto3 

96 

97 session = boto3.Session(region_name=s3.region) 

98 client = S3Client(local_cache_dir=cache_dir, file_cache_mode="persistent", boto3_session=session) 

99 else: 

100 client = S3Client(local_cache_dir=cache_dir, file_cache_mode="persistent") 

101 client.set_as_default_client() 

102 

103 

104def resolve_document(raw: str, cache_dir: Path | None = None) -> Path: 

105 """Coerce a user-supplied path string to a local filesystem Path. 

106 

107 Local inputs pass straight through. For ``s3://`` inputs, log an INFO line 

108 when the document is already present in the local cache (cache hit), then 

109 materialize it (cloudpathlib downloads on miss) and return the local Path. 

110 ``cache_dir`` defaults to ``get_settings().s3_cache.cache_dir``. 

111 

112 The cache-hit check mirrors cloudpathlib's on-disk layout 

113 (``<cache_dir>/<bucket>/<key>``) using only public ``S3Path`` attributes. 

114 Note: cloudpathlib still revalidates the S3 ETag on access, so a logged hit 

115 means "present locally"; if the source object changed upstream, cloudpathlib 

116 transparently re-downloads the fresh copy. 

117 """ 

118 doc = AnyPath(raw) 

119 if isinstance(doc, S3Path): 

120 cache_dir = cache_dir if cache_dir is not None else get_settings().s3_cache.cache_dir 

121 cached = cache_dir / doc.bucket / doc.key 

122 if cached.exists(): 

123 logger.info("S3 cache hit: {} served from {}", doc, cached) 

124 else: 

125 logger.debug("S3 cache miss: downloading {} to {}", doc, cached) 

126 return Path(doc) # __fspath__ materializes / revalidates an S3Path; no-op for a local Path 

127 

128 

129def fetch_file(src: str, dest: Path) -> Path: 

130 """Copy a local path or s3:// URI to `dest`, bypassing the cache. Returns `dest`. 

131 

132 Unlike `resolve_document`, the caller names the destination — used where a 

133 file must land at a specific staged path rather than wherever the cache 

134 keeps it. 

135 """ 

136 dest.parent.mkdir(parents=True, exist_ok=True) 

137 src_path = AnyPath(src) 

138 if isinstance(src_path, S3Path): 

139 logger.info("Downloading {src} -> {dest}", src=src, dest=dest) 

140 src_path.download_to(dest) 

141 else: 

142 dest.write_bytes(Path(src).read_bytes()) 

143 return dest 

144 

145 

146def resolve_artifacts(raw: str, filenames: Sequence[str], cache_dir: Path | None = None) -> Path: 

147 """Coerce an artifacts-directory string to a local directory Path. 

148 

149 A local directory passes straight through. For an ``s3://`` prefix, each 

150 named artifact file is materialized through the persistent cache — exactly 

151 ``resolve_document`` per file, with the same hit/miss logging and ETag 

152 revalidation — and the returned Path is the local cache directory that 

153 mirrors the prefix, so a caller written against a directory of files 

154 (``ParseResult.load``) reads the cached copies in place. 

155 

156 ``filenames`` must name every file the caller will read: only the named 

157 files are downloaded, nothing else under the prefix. 

158 """ 

159 if not filenames: 

160 raise ValueError("resolve_artifacts: filenames must name at least one artifact file") 

161 directory = AnyPath(raw) 

162 if not isinstance(directory, S3Path): 

163 return Path(raw) 

164 files = [resolve_document(str(directory / name), cache_dir) for name in filenames] 

165 return files[0].parent