MCPcopy Create free account
hub / github.com/TheSmallHanCat/flow2api / _write_cached_content

Method _write_cached_content

src/services/file_cache.py:166–178  ·  view source on GitHub ↗

先写临时文件,再原子替换,避免并发读到半截文件。

(self, file_path: Path, content: bytes)

Source from the content-addressed store, hash-verified

164 return headers
165
166 def _write_cached_content(self, file_path: Path, content: bytes):
167 """先写临时文件,再原子替换,避免并发读到半截文件。"""
168 temp_path = file_path.with_suffix(f"{file_path.suffix}.part")
169 try:
170 with open(temp_path, "wb") as f:
171 f.write(content)
172 temp_path.replace(file_path)
173 finally:
174 if temp_path.exists():
175 try:
176 temp_path.unlink()
177 except Exception:
178 pass
179
180 async def start_cleanup_task(self):
181 """Start background cleanup task"""

Callers 2

download_and_cacheMethod · 0.95
cache_base64_videoMethod · 0.95

Calls

no outgoing calls

Tested by

no test coverage detected