mirror of
https://github.com/OpenHands/OpenHands.git
synced 2026-03-22 13:47:19 +08:00
7edebcbc0c
Co-authored-by: openhands <openhands@all-hands.dev> Co-authored-by: OpenHands Bot <contact@all-hands.dev>
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
import os
|
|
import shutil
|
|
import threading
|
|
|
|
from openhands.core.logger import openhands_logger as logger
|
|
from openhands.storage.files import FileStore
|
|
|
|
|
|
class LocalFileStore(FileStore):
|
|
root: str
|
|
|
|
def __init__(self, root: str):
|
|
if root.startswith('~'):
|
|
root = os.path.expanduser(root)
|
|
self.root = root
|
|
os.makedirs(self.root, exist_ok=True)
|
|
|
|
def get_full_path(self, path: str) -> str:
|
|
if path.startswith('/'):
|
|
path = path[1:]
|
|
return os.path.join(self.root, path)
|
|
|
|
def write(self, path: str, contents: str | bytes) -> None:
|
|
full_path = self.get_full_path(path)
|
|
os.makedirs(os.path.dirname(full_path), exist_ok=True)
|
|
mode = 'w' if isinstance(contents, str) else 'wb'
|
|
|
|
# Use atomic write: write to temp file, then rename
|
|
# This prevents race conditions where concurrent writes could corrupt the file
|
|
temp_path = f'{full_path}.tmp.{os.getpid()}.{threading.get_ident()}'
|
|
try:
|
|
with open(temp_path, mode) as f:
|
|
f.write(contents)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(temp_path, full_path)
|
|
except Exception:
|
|
if os.path.exists(temp_path):
|
|
os.remove(temp_path)
|
|
raise
|
|
|
|
def read(self, path: str) -> str:
|
|
full_path = self.get_full_path(path)
|
|
with open(full_path, 'r') as f:
|
|
return f.read()
|
|
|
|
def list(self, path: str) -> list[str]:
|
|
full_path = self.get_full_path(path)
|
|
files = [os.path.join(path, f) for f in os.listdir(full_path)]
|
|
files = [f + '/' if os.path.isdir(self.get_full_path(f)) else f for f in files]
|
|
return files
|
|
|
|
def delete(self, path: str) -> None:
|
|
try:
|
|
full_path = self.get_full_path(path)
|
|
if not os.path.exists(full_path):
|
|
logger.debug(f'Local path does not exist: {full_path}')
|
|
return
|
|
if os.path.isfile(full_path):
|
|
os.remove(full_path)
|
|
logger.debug(f'Removed local file: {full_path}')
|
|
elif os.path.isdir(full_path):
|
|
shutil.rmtree(full_path)
|
|
logger.debug(f'Removed local directory: {full_path}')
|
|
except Exception as e:
|
|
logger.error(f'Error clearing local file store: {str(e)}')
|