File operations¶
Full Store API: read, write, delete, move, copy, list, metadata, type checks, capabilities, to_key.
"""File operations — Full Store API: read, write, delete, move, copy, list, metadata, type checks, capabilities, to_key.
Covers read/write, delete, move, copy, listing, metadata, and capability
checks — using the local backend.
---
see_also:
- label: Getting Started
url: ../../tutorial/getting-started.md
note: step-by-step guide
"""
from __future__ import annotations
import os
import tempfile
from remote_store import BackendConfig, Capability, Registry, RegistryConfig, Store, StoreProfile
def demo(store: Store) -> None:
"""Exercise core Store operations (everything except to_key)."""
# --- Write ---
result = store.write("docs/readme.txt", b"First file")
_ = store.write("docs/changelog.txt", b"v0.1.0 - initial release")
_ = store.write("data/report.csv", b"col1,col2\n1,2\n3,4")
_ = store.write("tmp/scratch.txt", b"temporary data")
print(f"Created 4 files (e.g. readme.txt: {result.size} bytes).\n")
# --- List files in a folder ---
print("Files in docs/:")
for f in store.list_files("docs"):
print(f" {f.name} ({f.size} bytes)")
# --- List folders at the store root ---
print("\nFolders at store root:")
for folder in store.list_folders(""):
print(f" {folder.name}/")
# --- Pattern-filtered folder listing ---
print("\nFolders matching 'd*':")
for folder in store.list_folders("", pattern="d*"):
print(f" {folder.name}/")
# --- Depth-limited folder listing (BFS traversal) ---
print("\nFolders up to 1 level deep:")
for folder in store.list_folders("", max_depth=1):
print(f" {folder.path}/")
# --- Read ---
content = store.read_bytes("data/report.csv")
print(f"\nreport.csv content:\n{content.decode()}")
# --- Metadata ---
info = store.get_file_info("docs/readme.txt")
print(f"readme.txt - size: {info.size}, modified: {info.modified_at}")
folder_info = store.get_folder_info("docs")
print(f"docs/ - {folder_info.file_count} files, {folder_info.total_size} bytes total")
# --- Depth-limited folder info (direct children only) ---
shallow = store.get_folder_info("", max_depth=0)
print(f"root (depth 0) - {shallow.file_count} files, {shallow.total_size} bytes")
full = store.get_folder_info("")
print(f"root (full) - {full.file_count} files, {full.total_size} bytes")
# --- is_file / is_folder ---
print(f"\nis_file('docs/readme.txt'): {store.is_file('docs/readme.txt')}")
print(f"is_folder('docs'): {store.is_folder('docs')}")
print(f"is_file('docs'): {store.is_file('docs')}")
print(f"is_folder('docs/readme.txt'):{store.is_folder('docs/readme.txt')}")
# --- Copy ---
store.copy("docs/readme.txt", "docs/readme_backup.txt")
print(f"\nCopied readme.txt -> readme_backup.txt (exists: {store.exists('docs/readme_backup.txt')})")
# --- Move ---
store.move("docs/changelog.txt", "archive/changelog.txt")
print(f"Moved changelog.txt -> archive/ (original exists: {store.exists('docs/changelog.txt')})")
# --- Delete ---
store.delete("docs/readme_backup.txt")
print(f"Deleted readme_backup.txt (exists: {store.exists('docs/readme_backup.txt')})")
# --- delete_folder ---
store.delete_folder("tmp", recursive=True)
print(f"Deleted tmp/ folder (exists: {store.exists('tmp')})")
# --- supports ---
print(f"\nBackend supports ATOMIC_WRITE: {store.supports(Capability.ATOMIC_WRITE)}")
print(f"Backend supports READ: {store.supports(Capability.READ)}")
# --- Recursive listing from store root ---
print("\nAll files (recursive):")
for f in store.list_files("", recursive=True):
print(f" {f.path} ({f.size} bytes)")
# --- Depth-limited file listing ---
print("\nFiles up to 1 level deep:")
for f in store.list_files("", max_depth=1):
print(f" {f.path} ({f.size} bytes)")
if __name__ == "__main__":
with tempfile.TemporaryDirectory() as tmp:
config = RegistryConfig(
backends={"local": BackendConfig(type="local", options={"root": tmp})},
stores={"files": StoreProfile(backend="local", root_path="workspace")},
)
with Registry(config) as registry:
store = registry.get_store("files")
demo(store)
# --- to_key (local-backend-specific) ---
absolute_path = os.path.join(tmp, "workspace", "data", "report.csv")
store_key = store.to_key(absolute_path)
print(f"\nto_key('{absolute_path}')")
print(f" -> '{store_key}'")
print("\nDone!")
See also¶
- Getting Started — step-by-step guide
- Source:
examples/getting_started/file_operations.py