alex
f925ae094c
- Abstract SourceAdapter with FileInfo dataclass - LocalAdapter for filesystem backups - Adapter factory for extensibility
48 lines
1.6 KiB
Python
48 lines
1.6 KiB
Python
import os
|
|
import aiofiles
|
|
from pathlib import Path
|
|
from typing import List, AsyncIterator, Dict, Any
|
|
from .base import SourceAdapter, FileInfo
|
|
|
|
class LocalAdapter(SourceAdapter):
|
|
async def connect(self) -> None:
|
|
base_path = self.config.get("path", ".")
|
|
if not os.path.exists(base_path):
|
|
raise FileNotFoundError(f"Path not found: {base_path}")
|
|
|
|
async def disconnect(self) -> None:
|
|
pass
|
|
|
|
async def list_files(self, path: str = "") -> List[FileInfo]:
|
|
base_path = Path(self.config.get("path", "."))
|
|
target_path = base_path / path if path else base_path
|
|
|
|
files = []
|
|
exclude_patterns = self.config.get("exclude", [])
|
|
|
|
for item in target_path.iterdir():
|
|
# Check exclude patterns
|
|
if any(item.match(pattern) for pattern in exclude_patterns):
|
|
continue
|
|
|
|
stat = item.stat()
|
|
files.append(FileInfo(
|
|
path=str(item.relative_to(base_path)),
|
|
size=stat.st_size,
|
|
modified_time=stat.st_mtime,
|
|
is_directory=item.is_dir()
|
|
))
|
|
|
|
return files
|
|
|
|
async def read_file(self, path: str) -> AsyncIterator[bytes]:
|
|
base_path = Path(self.config.get("path", "."))
|
|
file_path = base_path / path
|
|
|
|
async with aiofiles.open(file_path, "rb") as f:
|
|
while chunk := await f.read(8192):
|
|
yield chunk
|
|
|
|
async def get_database_dump(self, config: Dict[str, Any]) -> AsyncIterator[bytes]:
|
|
raise NotImplementedError("Local adapter does not support database dumps")
|