Update virtual_fs.py
This commit is contained in:
+42
-23
@@ -1,39 +1,58 @@
|
|||||||
"""
|
"""Simple in‑memory virtual file system.
|
||||||
Virtual file system used by the deep agent.
|
|
||||||
It stores files in memory and can be dumped to the real file system.
|
The VFS is just a dictionary mapping file names to string content.
|
||||||
|
It is shared between all tools and can be exported to the real
|
||||||
|
filesystem when requested.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import os
|
from __future__ import annotations
|
||||||
|
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Dict, Optional
|
from typing import Dict, Iterable
|
||||||
|
|
||||||
|
|
||||||
class VirtualFileSystem:
|
class VirtualFileSystem:
|
||||||
"""A simple in‑memory file system.
|
"""In‑memory file system.
|
||||||
|
|
||||||
Files are stored as a mapping from a relative path to its content.
|
Attributes
|
||||||
|
----------
|
||||||
|
files : Dict[str, str]
|
||||||
|
Mapping of file names to file content.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, base_dir: str = "virtual_fs"):
|
def __init__(self) -> None:
|
||||||
self.base_dir = Path(base_dir)
|
|
||||||
self.files: Dict[str, str] = {}
|
self.files: Dict[str, str] = {}
|
||||||
|
|
||||||
def write_file(self, path: str, content: str) -> None:
|
def create_file(self, name: str, content: str) -> None:
|
||||||
norm_path = Path(path).as_posix()
|
"""Create or overwrite a virtual file."""
|
||||||
self.files[norm_path] = content
|
self.files[name] = content
|
||||||
|
|
||||||
def read_file(self, path: str) -> Optional[str]:
|
def list_files(self) -> Iterable[str]:
|
||||||
norm_path = Path(path).as_posix()
|
"""Return an iterable of file names."""
|
||||||
return self.files.get(norm_path)
|
return self.files.keys()
|
||||||
|
|
||||||
def list_files(self) -> Dict[str, str]:
|
def read_file(self, name: str) -> str:
|
||||||
return dict(self.files)
|
"""Return the content of a virtual file.
|
||||||
|
|
||||||
def dump_to_real_fs(self, real_root: str = "real_fs") -> None:
|
Raises ``KeyError`` if the file does not exist.
|
||||||
real_root_path = Path(real_root)
|
"""
|
||||||
for rel_path, content in self.files.items():
|
return self.files[name]
|
||||||
file_path = real_root_path / rel_path
|
|
||||||
|
def export_to_disk(self, root: str | Path = ".") -> None:
|
||||||
|
"""Export all virtual files to the real filesystem.
|
||||||
|
|
||||||
|
Parameters
|
||||||
|
----------
|
||||||
|
root : str | Path, optional
|
||||||
|
Root directory where the virtual files will be written.
|
||||||
|
Defaults to the current working directory.
|
||||||
|
"""
|
||||||
|
root_path = Path(root).expanduser().resolve()
|
||||||
|
root_path.mkdir(parents=True, exist_ok=True)
|
||||||
|
for name, content in self.files.items():
|
||||||
|
file_path = root_path / name
|
||||||
file_path.parent.mkdir(parents=True, exist_ok=True)
|
file_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
file_path.write_text(content, encoding="utf-8")
|
file_path.write_text(content, encoding="utf-8")
|
||||||
|
|
||||||
def __repr__(self) -> str:
|
def __repr__(self) -> str: # pragma: no cover
|
||||||
return f"VirtualFileSystem(files={list(self.files.keys())})"
|
return f"VirtualFileSystem({len(self.files)} files)"
|
||||||
|
|||||||
Reference in New Issue
Block a user