Jan Lindemann
1e613a39c6
Add type annotations and casts to functions that were returning Any where a specific type was declared, satisfying the new warn_return_any mypy rule. Fixes: - log.py: get_caller_pos return type via cast - AsyncRunner.py: cast T for fut.result() - util.py: cast for getattr result, str() for args.username - FileContext.py: verbose_default bool annotation - SSHClient.py: cast SSHClient for dynamic import - lib/App.py: cast ArgumentParser, add return types to inner funcs - pm/rpm.py, dpkg.py: cast Iterable[Package] - App.py: cast for self.args.func(), add return types to inner funcs - BaseCmdPkgRelations.py: cast str for args.delimiter Assisted-by: unsloth/Qwen3.6-35B-A3B-GGUF:IQ4_NL and pi.dev 0.81.1 Signed-off-by: Jan Lindemann <jan@janware.com>
70 lines
2.2 KiB
Python
70 lines
2.2 KiB
Python
from typing import Any, Self
|
|
|
|
from .FileContext import FileContext
|
|
from .Uri import Uri
|
|
|
|
class CopyContext:
|
|
|
|
def __init__(
|
|
self,
|
|
src: Uri | str | FileContext,
|
|
dst: Uri | str | FileContext,
|
|
chroot: bool = False,
|
|
) -> None:
|
|
|
|
def __uri(ctx: FileContext | Uri | str) -> Uri | str:
|
|
if isinstance(ctx, Uri):
|
|
return ctx
|
|
if isinstance(ctx, str):
|
|
return ctx
|
|
assert isinstance(ctx, FileContext)
|
|
return ctx.uri
|
|
|
|
def __info(
|
|
ctx: FileContext | Uri | str,
|
|
) -> tuple[FileContext | None, str | Uri | None]:
|
|
fc: FileContext | None = ctx if isinstance(ctx, FileContext) else None
|
|
return fc, __uri(ctx)
|
|
|
|
self.__src, self.__src_uri = __info(src)
|
|
self.__dst, self.__dst_uri = __info(dst)
|
|
self.__chroot = chroot
|
|
|
|
async def __aenter__(self) -> Self:
|
|
if self.__src is None:
|
|
if self.__src_uri is None:
|
|
raise Exception('Tried to create source context without URI')
|
|
self.__src = FileContext.create(self.__src_uri, chroot = self.__chroot)
|
|
await self.__src.open()
|
|
if self.__dst is None:
|
|
if self.__dst_uri is None:
|
|
raise Exception('Tried to create destination context without URI')
|
|
self.__dst = FileContext.create(self.__dst_uri, chroot = self.__chroot)
|
|
await self.__dst.open()
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
|
|
if self.__src is not None:
|
|
await self.__src.close()
|
|
self.__src = None
|
|
if self.__dst is not None:
|
|
await self.__dst.close()
|
|
self.__dst = None
|
|
|
|
@property
|
|
def src(self) -> FileContext:
|
|
if self.__src is None:
|
|
raise Exception('Tried to access inexistent source context')
|
|
return self.__src
|
|
|
|
@property
|
|
def dst(self) -> FileContext:
|
|
if self.__dst is None:
|
|
raise Exception('Tried to access inexistent destination context')
|
|
return self.__dst
|
|
|
|
async def _run(self) -> None:
|
|
raise NotImplementedError('CopyContext._run() must be overridden')
|
|
|
|
async def run(self) -> None:
|
|
await self._run()
|