lib.FileContext: Add file methods
Add the following methods, meant to do the obvious:
unlink(self, path: str) -> None erase(self, path: str) -> None rename(self, src: str, dst: str) -> None mktemp(self, tmpl: str, directory: bool=False) -> None chown(self, path: str, owner: str|None=None, group: str|None=None) -> None chmod(self, path: str, mode: int) -> None stat(self, path: str, follow_symlinks: bool=True) -> StatResult file_exists(self, path: str) -> bool is_dir(self, path: str) -> bool
All methods are async and call their protected counterpart, which is designed to be overridden. If possible, default implementations do something meaningful, if not, they just raise plain NotImplementedError.
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
parent
bc3ed1737f
commit
3cf5b2264e
5 changed files with 295 additions and 10 deletions
|
|
@ -9,7 +9,7 @@ if TYPE_CHECKING:
|
|||
from typing import Self
|
||||
|
||||
from .log import *
|
||||
from .base import Input, InputMode, Result
|
||||
from .base import Input, InputMode, Result, StatResult
|
||||
|
||||
class FileContext(abc.ABC):
|
||||
|
||||
|
|
@ -109,6 +109,85 @@ class FileContext(abc.ABC):
|
|||
return await self._put(path, content, wd=wd, throw=throw, verbose=verbose,
|
||||
title=title, owner=owner, group=group, mode=mode_str, atomic=atomic)
|
||||
|
||||
async def _unlink(self, path: str) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: unlink() is not implemented')
|
||||
|
||||
async def unlink(self, path: str) -> None:
|
||||
return await self._unlink(path)
|
||||
|
||||
async def _erase(self, path: str) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: erase() is not implemented')
|
||||
|
||||
async def erase(self, path: str) -> None:
|
||||
return await self._erase(path)
|
||||
|
||||
async def _rename(self, src: str, dst: str) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: rename() is not implemented')
|
||||
|
||||
async def rename(self, src: str, dst: str) -> None:
|
||||
return await self._rename(src, dst)
|
||||
|
||||
async def _mktemp(self, tmpl: str, directory: bool) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: mktemp() is not implemented')
|
||||
|
||||
async def mktemp(self, tmpl: str, directory: bool=False) -> None:
|
||||
return await self._mktemp(tmpl, directory)
|
||||
|
||||
async def _chown(self, path: str, owner: str|None, group: str|None) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: chown() is not implemented')
|
||||
|
||||
async def chown(self, path: str, owner: str|None=None, group: str|None=None) -> None:
|
||||
if owner is None and group is None:
|
||||
raise ValueError(f'Tried to change ownership of {path} specifying neither owner nor group')
|
||||
return await self._chown(path, owner, group)
|
||||
|
||||
async def _chmod(self, path: str, mode: int) -> None:
|
||||
raise NotImplementedError(f'{self.log_name}: chmod() is not implemented')
|
||||
|
||||
async def chmod(self, path: str, mode: int) -> None:
|
||||
return await self._chmod(path, mode)
|
||||
|
||||
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
|
||||
raise NotImplementedError(f'{self.log_name}: lstat() is not implemented')
|
||||
|
||||
async def stat(self, path: str, follow_symlinks: bool=True) -> StatResult:
|
||||
if not isinstance(path, str):
|
||||
raise TypeError(f"path must be str, got {type(path).__name__}")
|
||||
return await self._stat(path, follow_symlinks)
|
||||
|
||||
async def _file_exists(self, path: str) -> bool:
|
||||
try:
|
||||
self._stat(path, False)
|
||||
except FileNotFoundError as e:
|
||||
log(DEBUG, f'Could not stat file {path} ({str(e)}), ignored')
|
||||
return False
|
||||
except Exception as e:
|
||||
log(ERR, f'Could not stat file {path} ({str(e)}), ignored')
|
||||
raise
|
||||
return True
|
||||
|
||||
async def file_exists(self, path: str) -> bool:
|
||||
return self._file_exists(path)
|
||||
|
||||
async def _is_dir(self, path: str) -> bool:
|
||||
try:
|
||||
return S_ISDIR(await self._stat(path).st_mode)
|
||||
except NotImplementedError:
|
||||
log(DEBUG, f'{self.log_name} doesn\'t implement stat(), judging by trailing slash if {path} is a directory')
|
||||
return path[-1] == '/'
|
||||
except FileNotFoundError as e:
|
||||
log(DEBUG, f'{self.log_name}: Failed to stat({path}) ({str(e)})')
|
||||
return False
|
||||
except Exception as e:
|
||||
log(ERR, f'{self.log_name}: Failed to stat({path}) ({str(e)})')
|
||||
raise
|
||||
return False
|
||||
|
||||
async def is_dir(self, path: str) -> bool:
|
||||
if not path:
|
||||
raise ValueError('Tried to investigate file system resource with empty path')
|
||||
return self._is_dir(path)
|
||||
|
||||
async def _close(self) -> None:
|
||||
pass
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue