App, lib, cmds: Fix mypy.explicit-override fallout

This commit adds @override decorators to approximately 300 methods
across 76 files that inherit from base classes such as AbstractCmd,
FileContext, ExecContext, Distro, SSHClient, and others.

The decorator ensures the type checker can verify that overridden
methods have compatible signatures and prevents accidental shadowing
of inherited methods without intent.

Files modified include command classes, library modules, distro
implementations, and SSH client implementations.

Assisted-by: unsloth/Qwen3.6-35B-A3B-GGUF:IQ4_NL with pi.dev v
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-07-22 15:51:14 +02:00
commit 5fa008be5a
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
78 changed files with 263 additions and 81 deletions

View file

@ -10,7 +10,7 @@ import sys
from enum import Enum, auto
from functools import cache
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from .lib.App import App as Base
from .lib.Distro import Distro
@ -374,11 +374,13 @@ class App(Base):
default_pkg_filter = pkg_filter,
)
@override
async def __aexit__(self, exc_type, exc, tb) -> None:
if self.__exec_context is not None:
await self.__exec_context.close()
self.__exec_context = None
@override
def _add_arguments(self, parser) -> None:
super()._add_arguments(parser)
parser.add_argument('-t', '--topdir', default = None, help = 'Project Path')
@ -421,6 +423,7 @@ class App(Base):
help = 'Default filter for all distribution package-related operations',
)
@override
async def _run(self, args: argparse.Namespace) -> None:
self.___topdir = args.topdir
self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format)

View file

@ -1,8 +1,10 @@
from typing import override
from .App import App
from .lib.Cmd import Cmd as Base
class CmdBase(Base):
@override # pyright: ignore[reportArgumentType]
@property
def app(self) -> App:
ret = super().app

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ..App import App as Parent
from ..CmdBase import CmdBase as Base
@ -20,6 +20,7 @@ class Cmd(Base): # export
) -> None:
super().__init__(parent, name, help, aliases = aliases)
@override
async def _run(self, args):
# Missing subcommand
self.print_help(1)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -18,6 +18,7 @@ class CmdPkg(Cmd): # export
)
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -25,5 +26,6 @@ class CmdPkg(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -15,6 +15,7 @@ class CmdPlatform(Cmd): # export
)
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -22,5 +23,6 @@ class CmdPlatform(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -20,6 +20,7 @@ class CmdPosix(Cmd): # export
)
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -27,5 +28,6 @@ class CmdPosix(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -17,6 +17,7 @@ class CmdProjects(Cmd): # export
)
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -24,5 +25,6 @@ class CmdProjects(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -13,6 +13,7 @@ class CmdSecrets(Cmd): # export
super().__init__(parent, 'secrets', help = 'Manage package secrets')
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -20,5 +21,6 @@ class CmdSecrets(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -12,11 +12,13 @@ class CmdDelete(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'delete', help = 'Delete packages by name')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
'names', nargs = '*', help = 'Names of packages to be deleted'
)
@override
async def _run(self, args: Namespace) -> None:
return await self.distro.delete(args.names)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -12,6 +12,7 @@ class CmdDup(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'dup', help = 'Upgrade distribution')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -21,5 +22,6 @@ class CmdDup(Cmd): # export
help = "Only download packages from the repos, don't install them, yet",
)
@override
async def _run(self, args: Namespace) -> None:
return await self.distro.dup(download_only = args.download_only)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from .Cmd import Cmd, Parent
@ -16,6 +16,7 @@ class CmdInstall(Cmd): # export
help = "Install the distribution's notion of available packages",
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('names', nargs = '*', help = 'Packages to be installed')
@ -32,6 +33,7 @@ class CmdInstall(Cmd): # export
help = "Don't expand macros in <names>",
)
@override
async def _run(self, args: Namespace) -> None:
names = (
args.names if args.fixed_strings else self.distro.expand_macros(args.names)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .NamedPkgsCmd import NamedPkgsCmd
@ -14,9 +14,11 @@ class CmdLs(NamedPkgsCmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'ls', help = 'List package contents')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
for name in args.names:
print('\n'.join(await self.distro.pkg_files(name)))

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .NamedPkgsCmd import NamedPkgsCmd, Parent
@ -12,9 +12,11 @@ class CmdMeta(NamedPkgsCmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'meta', help = 'List package metadata')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
names = await self.distro.select(args.names)
for name in names:

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -16,8 +16,10 @@ class CmdRebootRequired(Cmd): # export
help = 'Check whether the machine needs rebooting'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
await self.distro.reboot_required()

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -16,8 +16,10 @@ class CmdRefresh(Cmd): # export
help = "Refresh the distribution's notion of available packages",
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
return await self.distro.ref()

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...lib.PackageFilter import PackageFilterString
from .Cmd import Cmd, Parent
@ -13,10 +13,12 @@ class CmdSelect(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'select', help = 'Select packages by filter')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('filter', help = 'Package filter string')
@override
async def _run(self, args: Namespace) -> None:
filter = PackageFilterString(args.filter) if args.filter else None
for p in await self.distro.select(filter = filter):

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -12,6 +12,7 @@ class NamedPkgsCmd(Cmd): # export
def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('names', nargs = '*', help = 'Package names')

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...lib.Distro import Distro
from .Cmd import Cmd, Parent
@ -15,6 +15,7 @@ class CmdInfo(Cmd): # export
parent, 'info', help = 'Retrieve information about target platform'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -23,5 +24,6 @@ class CmdInfo(Cmd): # export
help = f'Format string, expanding macros {", ".join(Distro.macros())}',
)
@override
async def _run(self, args: Namespace) -> None:
print(self.app.distro.expand_macros(args.format))

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...CmdBase import CmdBase as Base
from ..CmdPosix import CmdPosix as Parent
@ -13,6 +13,7 @@ class Cmd(Base): # export
def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...lib.util import copy
from .Cmd import Cmd, Parent
@ -13,6 +13,7 @@ class CmdCopy(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'copy', help = 'Copy files')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('src', help = 'Source file URI')
@ -33,6 +34,7 @@ class CmdCopy(Cmd): # export
help = "Don't expand macros in <src> and <dst>",
)
@override
async def _run(self, args: Namespace) -> None:
def __expand(url: str) -> str:

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -13,6 +13,7 @@ class CmdTar(Cmd): # export
super().__init__(parent, 'tar', help = 'Handle tar archives')
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -20,5 +21,6 @@ class CmdTar(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ....CmdBase import CmdBase as Base
from ....lib.FileContext import FileContext
@ -27,6 +27,7 @@ class Cmd(Base): # export
)
yield ret
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ....lib.log import DEBUG, log
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,10 +12,12 @@ class CmdExtract(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'x', help = 'Extract a tar archive')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('dst', help = 'Destination root URI')
@override
async def _run(self, args: Namespace) -> None:
async with self.ctx(dst = args.dst) as ctx:
paths = await ctx.extract(ctx.dst.root)

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import re
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
from .lib.pkg_relations import VersionSyntax
@ -48,6 +48,7 @@ class BaseCmdPkgRelations(Cmd):
super().__init__(parent, 'pkg-' + relation, help = help)
self.relation = relation
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -141,6 +142,7 @@ class BaseCmdPkgRelations(Cmd):
help = 'Put double quotes around each listed dependency',
)
@override
async def _run(self, args: Namespace) -> None:
return self.print_pkg_relations(self.relation, args)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...CmdBase import CmdBase as Base
from ..CmdProjects import CmdProjects as Parent
@ -13,6 +13,7 @@ class Cmd(Base): # export
def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -5,7 +5,7 @@ import os
import re
from functools import lru_cache
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...App import Scope
from ...lib.log import DEBUG, ERR, NOTICE, log
@ -22,6 +22,7 @@ class CmdBuild(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'build', help = 'janware software project build tool')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -90,6 +91,7 @@ class CmdBuild(Cmd): # export
help = 'Modules to be built',
)
@override
async def _run(self, args: Namespace) -> None:
@lru_cache(maxsize = None)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...lib.base import InputMode
from ...lib.log import NOTICE, log
@ -20,6 +20,7 @@ class CmdCanonicalizeRemotes(Cmd): # export
parent, 'canonicalize-remotes', help = 'Streamline janware Git remotes'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -30,6 +31,7 @@ class CmdCanonicalizeRemotes(Cmd): # export
help = 'Only log what would be done',
)
@override
async def _run(self, args: Namespace) -> None:
async def git(cmd: list[str], ro = False, throw = True) -> Result:

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...App import Scope
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,10 +12,12 @@ class CmdCflags(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'cflags', help = 'cflags')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser
@ -16,6 +16,7 @@ class CmdCheck(Cmd): # export
)
self.load_subcommands()
@override
async def _run(self, args):
import sys
@ -23,5 +24,6 @@ class CmdCheck(Cmd): # export
self.parser.print_help()
sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -11,9 +11,11 @@ class CmdCommands(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'commands', help = 'List available commands')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
import glob
import os

View file

@ -1,5 +1,6 @@
from argparse import ArgumentParser, ArgumentTypeError, Namespace
from enum import Enum, auto
from typing import override
from ...lib.log import WARNING, log
from .Cmd import Cmd, Parent
@ -93,6 +94,7 @@ class CmdCreateFile(Cmd): # export
parent, 'create-file', help = 'Generate a file from project metadata'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -124,6 +126,7 @@ class CmdCreateFile(Cmd): # export
)
parser.add_argument('module', help = 'The module to generate the file for')
@override
async def _run(self, args: Namespace) -> None:
method = getattr(self, 'render_' + args.format, None)
if method is None: # Should be prevented by choices=[] but keeps linter happy

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
from .lib.templates import tmpl_render
@ -32,6 +32,7 @@ class CmdCreatePkgConfig(Cmd): # export
string = re.sub(patt, replacement, string)
return string
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('-F', '--project-descr-file', default = None)
@ -46,6 +47,7 @@ class CmdCreatePkgConfig(Cmd): # export
parser.add_argument('-R', '--requires-build', default = None)
parser.add_argument('-V', '--variables', nargs = '*')
@override
async def _run(self, args: Namespace) -> None:
project_conf_var_keys = [
'description',

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...App import Scope
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,6 +12,7 @@ class CmdExepath(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'exepath', help = 'exepath')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -23,6 +24,7 @@ class CmdExepath(Cmd): # export
)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -5,7 +5,7 @@ import re
from ...lib.log import DEBUG, log
from ...lib.Uri import Uri
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -19,6 +19,7 @@ class CmdGetAuthInfo(Cmd): # export
help = 'Try to retrieve authentication information from the source tree',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -52,6 +53,7 @@ class CmdGetAuthInfo(Cmd): # export
help = 'Show remote base URL',
)
@override
async def _run(self, args: Namespace) -> None:
keys = ['username', 'password']

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -11,6 +11,7 @@ class CmdGetval(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'getval', help = 'Get value from project config')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -21,6 +22,7 @@ class CmdGetval(Cmd): # export
parser.add_argument('section', default = '', help = 'Config section')
parser.add_argument('key', default = '', help = 'Config key')
@override
async def _run(self, args: Namespace) -> None:
project = args.project
if project is None:

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -15,10 +15,12 @@ class CmdHtdocsDir(Cmd): # export
help = 'Print source directory containing document root of a given module',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
r = []
for m in args.module:

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...App import Scope
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,6 +12,7 @@ class CmdLdflags(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'ldflags', help = 'ldflags')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@ -56,6 +57,7 @@ class CmdLdflags(Cmd): # export
return None
return ' '.join(ret)
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...App import Scope
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,6 +12,7 @@ class CmdLdlibpath(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'ldlibpath', help = 'ldlibpath')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -23,6 +24,7 @@ class CmdLdlibpath(Cmd): # export
)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -11,9 +11,11 @@ class CmdLibname(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'libname', help = 'libname')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
print(self.app.get_libname(args.module))

View file

@ -6,7 +6,7 @@ from ...lib.log import DEBUG, log
from ...lib.Uri import Uri
from ...lib.util import get_password, get_username, run_curl_into
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -18,6 +18,7 @@ class CmdListRepos(Cmd): # export
parent, 'list-repos', help = 'Query a remote GIT server for repositories'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -43,6 +44,7 @@ class CmdListRepos(Cmd): # export
'--from-owner', help = "List from-owner's projects", default = 'janware'
)
@override
async def _run(self, args: Namespace) -> None:
base_url = Uri(args.base_url)

View file

@ -3,7 +3,7 @@ import re
from ...lib.log import DEBUG, log
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -13,6 +13,7 @@ class CmdModules(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'modules', help = 'Query existing janware packages')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -24,6 +25,7 @@ class CmdModules(Cmd): # export
'Key-value pairs, seperated by commas, to be searched for in project.conf',
)
@override
async def _run(self, args: Namespace) -> None:
import pathlib

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...App import Scope
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -12,10 +12,12 @@ class CmdPath(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'path', help = 'path')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -2,7 +2,7 @@ from __future__ import annotations
from ...lib.log import WARNING, log
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -14,10 +14,12 @@ class CmdProjDir(Cmd): # export
parent, 'proj-dir', help = 'Print directory of a given package'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
out = []
for m in args.module:

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ...App import Scope
from .Cmd import Cmd, Parent
@ -15,6 +15,7 @@ class CmdPythonpath(Cmd): # export
parent, 'pythonpath', help = 'Generate PYTHONPATH for given modules'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -38,6 +39,7 @@ class CmdPythonpath(Cmd): # export
)
parser.add_argument('module', help = 'Modules', nargs = '*')
@override
async def _run(self, args: Namespace) -> None:
deps = self.app.get_project_refs(
args.module,

View file

@ -3,7 +3,7 @@ from __future__ import annotations
from ...App import Scope
from ...lib.log import DEBUG, log
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -18,6 +18,7 @@ class CmdRequiredOsPkg(Cmd): # export
help = 'List distribution packages required for a package',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('flavours', help = 'Dependency flavours', default = 'build')
@ -35,6 +36,7 @@ class CmdRequiredOsPkg(Cmd): # export
help = 'Put double quotes around each listed dependency',
)
@override
async def _run(self, args: Namespace) -> None:
modules = args.modules
flavours = set(args.flavours.split(','))

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -13,10 +13,12 @@ class CmdSummary(Cmd): # export
parent, 'summary', help = 'Print summary description of given modules'
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
r = []
for m in args.module:

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -11,9 +11,11 @@ class CmdTest(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'test', help = 'Test')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('blah', default = '', help = 'The blah argument')
@override
async def _run(self, args: Namespace) -> None:
print('blah = ' + args.blah)

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace
@ -15,10 +15,12 @@ class CmdTmplDir(Cmd): # export
help = 'Print directory containing templates of a given module',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None:
r = []
for m in args.module:

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ....CmdBase import CmdBase as Base
from ..CmdCheck import CmdCheck as Parent
@ -13,6 +13,7 @@ class Cmd(Base): # export
def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ....lib.log import NOTICE, log
from .Cmd import Cmd, Parent
@ -17,11 +17,13 @@ class CmdDep(Cmd): # export
help = 'Check for circular dependencies between given modules',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules')
parser.add_argument('-f', '--flavour', nargs = '?', default = 'build')
@override
async def _run(self, args: Namespace) -> None:
cycle = self.app.find_circular_deps(args.module, args.flavour)
if cycle:

View file

@ -1,7 +1,7 @@
from __future__ import annotations
from functools import cached_property
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...CmdBase import CmdBase as Base
from ..CmdSecrets import CmdSecrets as Parent
@ -46,6 +46,7 @@ class Cmd(Base): # export
def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument('packages', nargs = '*', help = 'Package names')

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
from .lib.base import Attrs
@ -15,10 +15,12 @@ class CmdCompileTemplates(Cmd): # export
parent, 'compile-templates', help = 'Compile package template files'
)
@override
async def _run(self, args: Namespace) -> None:
attrs = Attrs(args.mode, args.owner, args.group, None)
await self._compile_template_files(args.packages, attrs)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -18,6 +18,7 @@ class CmdInstall(Cmd): # export
),
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
parser.add_argument('src', help = 'URI of secret source')
parser.add_argument(
@ -28,6 +29,7 @@ class CmdInstall(Cmd): # export
)
super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None:
packages = [] if args.packages == ['all'] else args.packages
await self.ctx.install(args.src, packages, args.only_missing)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -16,6 +16,7 @@ class CmdListCompilationOutput(Cmd): # export
help = 'List package compilation output files',
)
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -25,6 +26,7 @@ class CmdListCompilationOutput(Cmd): # export
help = 'Show all output targets, including non-existent files',
)
@override
async def _run(self, args: Namespace) -> None:
print(
'\n'.join(

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -12,6 +12,7 @@ class CmdListSecrets(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'list-secrets', help = 'List package secret files')
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(
@ -21,6 +22,7 @@ class CmdListSecrets(Cmd): # export
help = 'Show all secret paths, including non-existent files',
)
@override
async def _run(self, args: Namespace) -> None:
print(
'\n'.join(

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -12,5 +12,6 @@ class CmdListTemplates(Cmd): # export
def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'list-templates', help = 'List package template files')
@override
async def _run(self, args: Namespace) -> None:
print('\n'.join(await self._list_template_files(args.packages)))

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent
@ -16,5 +16,6 @@ class CmdRmCompilationOutput(Cmd): # export
help = 'Remove package compilation output files',
)
@override
async def _run(self, args: Namespace) -> None:
await self._remove_compilation_targets(args.packages)

View file

@ -6,7 +6,7 @@ import os
import sys
from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any
from typing import override, TYPE_CHECKING, Any
from .AsyncRunner import AsyncRunner
from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level
@ -192,6 +192,7 @@ class App: # export
class NoopCompleter(BaseCompleter):
@override
def __call__(self, *args, **kwargs):
return None

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import abc
import sys
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .log import ERR
from .Types import LoadTypes, Types
@ -173,14 +173,18 @@ class Cmd(AbstractCmd): # export
self.__description = description if description else help
self.__aliases = aliases if aliases else []
@override
def _name(self) -> str:
return self.__name
@override
def _help(self) -> str:
return self.__help
@override
def _description(self) -> str:
return self.__description
@override
def _aliases(self) -> Iterable[str]:
return self.__aliases

View file

@ -5,7 +5,7 @@ import errno
import sys
from decimal import ROUND_FLOOR, Decimal
from typing import TYPE_CHECKING, NamedTuple
from typing import override, TYPE_CHECKING, NamedTuple
if TYPE_CHECKING:
from typing import Type
@ -250,6 +250,7 @@ class ExecContext(Base):
super().__init__(*args, **kwargs)
@classmethod
@override
def create(cls, *args, **kwargs) -> ExecContext:
ret = super().create(*args, **kwargs)
if not isinstance(ret, cls):
@ -451,6 +452,7 @@ class ExecContext(Base):
cc.check_exit_code(ret)
return ret
@override
async def _get(
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result:
@ -485,6 +487,7 @@ class ExecContext(Base):
cc.check_exit_code(ret)
return ret
@override
async def _put(
self,
path: str,
@ -565,22 +568,27 @@ class ExecContext(Base):
log(ERR, msg)
return ret
@override
async def _unlink(self, path: str) -> None:
cmd = ['rm', '-f', path]
await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _erase(self, path: str) -> None:
cmd = ['rm', '-rf', path]
await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _rename(self, src: str, dst: str) -> None:
cmd = ['mv', src, dst]
await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _mkdir(self, path: str, mode: int) -> None:
cmd = ['mkdir', path, '-m', self.__mode_str(mode)]
await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _mktemp(self, tmpl: str, directory: bool) -> str:
cmd = ['mktemp']
if directory:
@ -593,6 +601,7 @@ class ExecContext(Base):
)
return result.stdout_str.strip()
@override
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
async def __stat(opts: list[str]) -> Result:
@ -662,6 +671,7 @@ class ExecContext(Base):
assert stdout is not None # Just there to pacify the linter
return _build_stat_result(stdout.rstrip('\n').split(_US), mode_base = 8)
@override
async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
if owner is None and group is None:
raise ValueError(f'Tried to chown("{path}") without owner and group')
@ -674,6 +684,7 @@ class ExecContext(Base):
assert ownership is not None # Impossible, just there to calm the linter
await self.run(['chown', ownership, path], cmd_input = InputMode.NonInteractive)
@override
async def _chmod(self, path: str, mode: int) -> None:
await self.run(
['chmod', self.__mode_str(mode), path],

View file

@ -4,7 +4,7 @@ import abc
from enum import Enum, auto
from functools import cached_property
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from .log import DEBUG, ERR, log
from .Uri import Uri
@ -50,6 +50,7 @@ class FileContext(abc.ABC):
async def __aexit__(self, exc_type, exc, tb):
await self.close()
@override
def __repr__(self) -> str:
return self.__uri.id

View file

@ -1,4 +1,4 @@
from typing import Any
from typing import override, Any
meta_tags = [
'name',
@ -56,6 +56,7 @@ class Package:
self.url = url
self.maintainer = maintainer
@override
def __repr__(self) -> str:
global meta_tags
return '\n'.join([f'{key:<15}: {getattr(self, key)}' for key in meta_tags])

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import abc
import re
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from .Package import Package
@ -25,6 +25,7 @@ class PackageFilterString(PackageFilter):
raise Exception(f'Unsupported filter string "{definition}"')
self.__definition = url_rx_str
@override
def _match(self, package: Package) -> bool:
url = package.url
if url is None:

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import abc
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .base import Result
@ -20,6 +20,7 @@ class ProcFilter(abc.ABC):
class ProcFilterIdentity(ProcFilter):
@override
async def _run(self, data: bytes | None) -> Result:
return Result(data, None, 0)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from .ProcFilter import ProcFilter
@ -13,6 +13,7 @@ class ProcFilterGpg(ProcFilter):
def __init__(self, ec: ExecContext) -> None:
self.__ec = ec
@override
async def _run(self, data: bytes | None) -> Result:
if data is None:
raise Exception('No data for GPG to decrypt')

View file

@ -1,5 +1,7 @@
from __future__ import annotations
from typing import override
class Result:
def __init__(
@ -113,6 +115,7 @@ class Result:
return ret
@override
def __repr__(self) -> str:
return self.__summarize(verbose = False)

View file

@ -9,7 +9,7 @@ from tarfile import TarFile, TarInfo
from .CopyContext import CopyContext
from .ExecContext import ExecContext
from .log import DEBUG, ERR, log
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
if TYPE_CHECKING:
from .base import StatResult
@ -89,6 +89,7 @@ class TarIo(CopyContext):
class TarIoTarFile(TarIo):
@override
async def _extract(self, blob: bytes, root: str | None = None) -> None:
tf = TarFile(fileobj = io.BytesIO(blob))
for info in tf.getmembers():
@ -111,7 +112,16 @@ class TarIoTarFile(TarIo):
class TarIoTarExec(TarIo):
@override
async def _extract(self, blob: bytes, root: str | None = None) -> None:
cmd = ['tar']
if root is not None:
cmd += ['-C', root]
cmd += ['-x', '-f', '-']
await self.dst.run(cmd, cmd_input = blob)
@property
@override
def dst(self) -> ExecContext:
ret = super().dst
if not isinstance(ret, ExecContext):
@ -120,10 +130,3 @@ class TarIoTarExec(TarIo):
'context, which only has a file context'
)
return ret
async def _extract(self, blob: bytes, root: str | None = None) -> None:
cmd = ['tar']
if root is not None:
cmd += ['-C', root]
cmd += ['-x', '-f', '-']
await self.dst.run(cmd, cmd_input = blob)

View file

@ -5,7 +5,7 @@ import os
import re
import sys
from typing import TYPE_CHECKING, Generic, Iterable, TypeVar
from typing import override, TYPE_CHECKING, Generic, Iterable, TypeVar
from .log import ERR, OFF, log, parse_log_level
@ -17,6 +17,7 @@ T = TypeVar('T')
class Types(abc.ABC, Iterable[type[T]], Generic[T]): # export
@override
def __iter__(self) -> Iterator[type[T]]:
return iter(self._classes())
@ -64,6 +65,7 @@ class LoadTypes(Types[T]): # export
if self.__debug_level != OFF:
log(self.__debug_level, *args, **kwargs)
@override
def _stringify(self):
tf = 'None' if self.__type_filter is None else (
', '.join([str(f) for f in self.__type_filter])
@ -74,6 +76,7 @@ class LoadTypes(Types[T]): # export
'mod_names: ' + ', '.join(self.__mod_names),
]
@override
def _classes(self) -> Iterable[type[T]]:
if self.__classes is None:

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import copy
from functools import cached_property
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
if TYPE_CHECKING:
import urllib.parse
@ -39,9 +39,11 @@ class Uri:
self.__username: str | None = None
self.__password: str | None = None
@override
def __repr__(self) -> str:
return self.full
@override
def __str__(self) -> str:
return self.safe_full_with_username

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...Distro import Distro as Base
@ -26,27 +26,32 @@ class Distro(Base):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@override
async def _ref(self) -> None:
raise NotImplementedError(
'distro refresh is not yet implemented for Arch-like distributions'
)
@override
async def _dup(self, download_only: bool) -> None:
args = ['-Su']
if download_only:
args.append('-w')
await self.pacman(args)
@override
async def _reboot_required(self, verbose: bool) -> bool:
raise NotImplementedError(
'distro reboot-required is not yet implemented for Arch-like distributions'
)
@override
async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]:
raise NotImplementedError(
'distro select is not yet implemented for Arch-like distributions'
)
@override
async def _install(self, names: Iterable[str], only_update: bool) -> None:
if only_update:
raise NotImplementedError('--only-update is not yet implemented for pacman')
@ -54,11 +59,13 @@ class Distro(Base):
args.extend(names)
await self.pacman(args)
@override
async def _delete(self, names: Iterable[str]) -> None:
raise NotImplementedError(
'distro delete not yet implemented for Arch-like distributions'
)
@override
async def _pkg_files(self, name: str) -> Iterable[str]:
raise NotImplementedError(
'distro pkg ls yet implemented for Arch-like distributions'

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import os
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...Distro import Distro as Base
from ...log import NOTICE, log
@ -38,9 +38,11 @@ class Distro(Base):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@override
async def _ref(self) -> None:
await self.apt_get(['update'])
@override
async def _dup(self, download_only: bool) -> None:
args: list[str] = []
if download_only:
@ -48,6 +50,7 @@ class Distro(Base):
args.append('upgrade')
await self.apt_get(args)
@override
async def _reboot_required(self, verbose: bool) -> bool:
reboot_required = '/run/reboot_required'
if os.path.exists(reboot_required):
@ -64,9 +67,11 @@ class Distro(Base):
log(NOTICE, f"No. {reboot_required} doesn't exist.")
return False
@override
async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]:
return await query_packages(names, ec = self.ctx)
@override
async def _install(self, names: Iterable[str], only_update: bool) -> None:
args = ['install']
if only_update:
@ -75,8 +80,10 @@ class Distro(Base):
args.extend(names)
await self.apt_get(args)
@override
async def _delete(self, names: Iterable[str]) -> None:
await self.dpkg(['-P', *names], sudo = True)
@override
async def _pkg_files(self, name: str) -> Iterable[str]:
return await list_files(name, ec = self.ctx)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...Distro import Distro as Base
from ...pm.rpm import list_files, query_packages, run_rpm
@ -43,9 +43,11 @@ class Distro(Base):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
@override
async def _ref(self) -> None:
await self.zypper(['refresh'])
@override
async def _dup(self, download_only: bool) -> None:
args = ['dup']
if download_only:
@ -53,6 +55,7 @@ class Distro(Base):
args += self.__update_extra_opts
await self.zypper(args)
@override
async def _reboot_required(self, verbose: bool) -> bool:
opts = []
if not verbose:
@ -64,22 +67,27 @@ class Distro(Base):
return True
return False
@override
async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]:
return await query_packages(names, ec = self.ctx)
@override
async def _install(self, names: Iterable[str], only_update: bool) -> None:
cmd = ['update' if only_update else 'install']
cmd += self.__update_extra_opts
cmd += names
await self.zypper(cmd)
@override
async def _install_local_files(
self, paths: Iterable[str], only_update: bool
) -> None:
await self.rpm(['-U', '--reinstall', *paths])
@override
async def _delete(self, names: Iterable[str]) -> None:
await self.rpm(['-e', *names], sudo = True)
@override
async def _pkg_files(self, name: str) -> Iterable[str]:
return await list_files(name, ec = self.ctx)

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ..FileContext import FileContext as Base
@ -28,6 +28,7 @@ class Curl(Base):
self.__ec = ec if ec else __local()
@override
async def _get(
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result:

View file

@ -7,7 +7,7 @@ import pwd
import sys
from functools import cache
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ..base import Result, StatResult
from ..ExecContext import ExecContext as Base
@ -25,6 +25,7 @@ class Local(Base):
def _username(self) -> str:
return pwd.getpwuid(os.getuid()).pw_name
@override
async def _run(
self,
cmd: list[str],
@ -155,9 +156,11 @@ class Local(Base):
if cwd is not None:
os.chdir(cwd)
@override
async def _unlink(self, path: str) -> None:
os.unlink(path)
@override
async def _erase(self, path: str) -> None:
if os.path.isdir(path):
import shutil
@ -165,26 +168,33 @@ class Local(Base):
return
os.unlink(path)
@override
async def _rename(self, src: str, dst: str) -> None:
os.rename(src, dst)
@override
async def _mkdir(self, path: str, mode: int) -> None:
os.mkdir(path, mode)
@override
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
return StatResult.from_os(os.stat(path, follow_symlinks = follow_symlinks))
@override
async def _file_exists(self, path: str) -> bool:
return os.path.exists(path)
@override
async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
uid = pwd.getpwnam(owner).pw_uid if owner else -1
gid = grp.getgrnam(group).gr_gid if group else -1
os.chown(path, uid, gid)
@override
async def _chmod(self, path: str, mode: int) -> None:
os.chmod(path, mode)
@override
async def _is_dir(self, path: str, follow_symlinks: bool) -> bool:
if (not follow_symlinks) and os.path.islink(path):
return False

View file

@ -5,7 +5,7 @@ import os
import pwd
from enum import Flag, auto
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, override
from ..ExecContext import ExecContext
from ..log import DEBUG, ERR, INFO, NOTICE, get_caller_pos, log, log_m
@ -42,6 +42,7 @@ class SSHClient(ExecContext):
) -> Result:
pass
@override
async def _run(
self,
cmd: list[str],
@ -102,6 +103,7 @@ class SSHClient(ExecContext):
return self.uri.port
@property
@override
def username(self) -> str | None:
return self.uri.username

View file

@ -1,3 +1,4 @@
from typing import override
import asyncio
import os
import shlex
@ -42,10 +43,12 @@ class AsyncSSH(Base):
self.__connect_timeout = connect_timeout
self.__conn: asyncssh.SSHClientConnection | None = None
@override
async def _open(self) -> None:
await super()._open()
await self._conn
@override
async def _close(self) -> None:
if self.__conn is not None:
try:
@ -355,6 +358,7 @@ class AsyncSSH(Base):
stdout = b''.join(stdout_parts) if stdout_parts else None
return Result(stdout, None, exit_code, cmd = cmd)
@override
async def _run_ssh(
self,
cmd: list[str],

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import os
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
from ...base import InputMode
from ...util import run_cmd
@ -48,6 +48,7 @@ class Exec(Base):
self.__askpass_orig[key] = os.getenv(key)
os.environ[key] = val
@override
async def _run_ssh(
self,
cmd: list[str],

View file

@ -1,6 +1,6 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from typing import override, TYPE_CHECKING
# Tolerate missing paramiko imports. jw-pkg is designed to work with what it
# finds.
@ -52,15 +52,18 @@ class Paramiko(Base):
def __scp(self) -> Any:
return paramiko.SCPClient(self.__client.get_transport())
@override
async def _open(self) -> None:
await super()._open()
self.__client
@override
async def _close(self) -> None:
if self.___client is not None:
self.___client.close()
self.___client = None
@override
async def _run_ssh(
self,
cmd: list[str],