Local
local
¶
Local file storage implementation
LocalFileClient(base_path: Path | str = '.')
¶
Local file storage client supporting various formats
This client stores data as local files and supports: - JSON files for general data storage - CSV/Excel files for spreadsheet-like data
Parameters:
Name | Type | Description | Default |
---|---|---|---|
base_path | Path | str | Base directory for storing files | '.' |
Source code in src/pytanis/storage/local.py
def __init__(self, base_path: Path | str = '.'):
"""Initialize the local file client
Args:
base_path: Base directory for storing files
"""
self.base_path = Path(base_path)
self.base_path.mkdir(parents=True, exist_ok=True)
base_path = Path(base_path)
instance-attribute
¶
create_spreadsheet(name: str) -> str
¶
Create a new empty spreadsheet
Source code in src/pytanis/storage/local.py
def create_spreadsheet(self, name: str) -> str:
"""Create a new empty spreadsheet"""
if not name.endswith(('.xlsx', '.csv')):
name = f'{name}.xlsx'
path = self._get_path(name)
if path.exists():
raise OSError(f'Spreadsheet already exists: {name}')
# Create empty file
if path.suffix == '.csv':
pd.DataFrame().to_csv(path, index=False)
else:
with pd.ExcelWriter(path, mode='w') as writer:
pd.DataFrame().to_excel(writer, sheet_name='Sheet1', index=False)
return name
delete(key: str) -> None
¶
Delete a file
Source code in src/pytanis/storage/local.py
def delete(self, key: str) -> None:
"""Delete a file"""
path = self._get_path(key)
if not path.exists():
raise KeyError(f'File not found: {key}')
try:
path.unlink()
except Exception as e:
raise OSError(f'Error deleting file {key}: {e}') from e
delete_sheet(spreadsheet_id: str, sheet_name: str) -> None
¶
Delete a sheet from a spreadsheet
Source code in src/pytanis/storage/local.py
def delete_sheet(self, spreadsheet_id: str, sheet_name: str) -> None:
"""Delete a sheet from a spreadsheet"""
path = self._get_spreadsheet_path(spreadsheet_id)
if not path.exists():
raise KeyError(f'Spreadsheet not found: {spreadsheet_id}')
if path.suffix == '.csv':
raise OSError('Cannot delete sheets from CSV files')
try:
# Read all sheets except the one to delete
excel_file = pd.ExcelFile(path)
sheets_to_keep = {
name: pd.read_excel(excel_file, sheet_name=name)
for name in excel_file.sheet_names
if name != sheet_name
}
if sheet_name not in excel_file.sheet_names:
raise KeyError(f'Sheet not found: {sheet_name}')
if not sheets_to_keep:
raise OSError('Cannot delete the last sheet in a spreadsheet')
# Write back remaining sheets
with pd.ExcelWriter(path, mode='w') as writer:
for name, df in sheets_to_keep.items():
df.to_excel(writer, sheet_name=str(name), index=False)
except Exception as e:
if isinstance(e, KeyError | OSError):
raise
raise OSError(f'Error deleting sheet {sheet_name} from {spreadsheet_id}: {e}') from e
exists(key: str) -> bool
¶
Check if a file exists
Source code in src/pytanis/storage/local.py
def exists(self, key: str) -> bool:
"""Check if a file exists"""
return self._get_path(key).exists()
list_keys(prefix: str = '') -> list[str]
¶
List all files with optional prefix filtering
Source code in src/pytanis/storage/local.py
def list_keys(self, prefix: str = '') -> list[str]:
"""List all files with optional prefix filtering"""
pattern = f'{prefix}*' if prefix else '*'
paths = self.base_path.rglob(pattern)
return [str(p.relative_to(self.base_path)) for p in paths if p.is_file()]
list_sheets(spreadsheet_id: str) -> list[str]
¶
List all sheets in a spreadsheet
Source code in src/pytanis/storage/local.py
def list_sheets(self, spreadsheet_id: str) -> list[str]:
"""List all sheets in a spreadsheet"""
path = self._get_spreadsheet_path(spreadsheet_id)
if not path.exists():
raise KeyError(f'Spreadsheet not found: {spreadsheet_id}')
if path.suffix == '.csv':
return ['default'] # CSV files don't have multiple sheets
try:
# For Excel files, read sheet names
excel_file = pd.ExcelFile(path)
# Convert all sheet names to strings
return [str(name) for name in excel_file.sheet_names]
except Exception as e:
raise OSError(f'Error listing sheets in {spreadsheet_id}: {e}') from e
read(key: str) -> Any
¶
Read data from a JSON file
Source code in src/pytanis/storage/local.py
def read(self, key: str) -> Any:
"""Read data from a JSON file"""
path = self._get_path(key)
if not path.exists():
raise KeyError(f'File not found: {key}')
try:
with open(path, encoding='utf-8') as f:
return json.load(f)
except Exception as e:
raise OSError(f'Error reading file {key}: {e}') from e
read_sheet(spreadsheet_id: str, sheet_name: str | None = None) -> pd.DataFrame
¶
Read a sheet from a CSV or Excel file
Source code in src/pytanis/storage/local.py
def read_sheet(self, spreadsheet_id: str, sheet_name: str | None = None) -> pd.DataFrame:
"""Read a sheet from a CSV or Excel file"""
path = self._get_spreadsheet_path(spreadsheet_id)
if not path.exists():
raise KeyError(f'Spreadsheet not found: {spreadsheet_id}')
try:
if path.suffix == '.csv':
if sheet_name is not None:
_logger.warning('Sheet name ignored for CSV files', sheet_name=sheet_name)
return pd.read_csv(path)
else: # Excel
# Always specify sheet_name to ensure we get a DataFrame, not a dict
sheet_to_read: str | int = sheet_name if sheet_name is not None else 0
return pd.read_excel(path, sheet_name=sheet_to_read)
except Exception as e:
raise OSError(f'Error reading spreadsheet {spreadsheet_id}: {e}') from e
write(key: str, data: Any) -> None
¶
Write data to a JSON file
Source code in src/pytanis/storage/local.py
def write(self, key: str, data: Any) -> None:
"""Write data to a JSON file"""
path = self._get_path(key)
try:
with open(path, 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2, default=str)
except Exception as e:
raise OSError(f'Error writing file {key}: {e}') from e
write_sheet(spreadsheet_id: str, data: pd.DataFrame, sheet_name: str | None = None, *, overwrite: bool = True) -> None
¶
Write a DataFrame to a CSV or Excel file
Source code in src/pytanis/storage/local.py
def write_sheet(
self, spreadsheet_id: str, data: pd.DataFrame, sheet_name: str | None = None, *, overwrite: bool = True
) -> None:
"""Write a DataFrame to a CSV or Excel file"""
path = self._get_spreadsheet_path(spreadsheet_id)
try:
if path.suffix == '.csv':
if sheet_name is not None:
_logger.warning('Sheet name ignored for CSV files', sheet_name=sheet_name)
data.to_csv(path, index=False)
elif path.exists() and not overwrite:
# Append to existing Excel file
with pd.ExcelWriter(path, mode='a', if_sheet_exists='replace') as writer:
data.to_excel(writer, sheet_name=sheet_name or 'Sheet1', index=False)
else:
# Create new or overwrite
with pd.ExcelWriter(path, mode='w') as writer:
data.to_excel(writer, sheet_name=sheet_name or 'Sheet1', index=False)
except Exception as e:
raise OSError(f'Error writing spreadsheet {spreadsheet_id}: {e}') from e