jw.pkg: Massively tighten linter checks #53

Merged
Jan Lindemann merged 31 commits from jan/feature/20260809-pyrightconfig-json-mirror-strict-mypy-settings into master 2026-08-09 15:12:37 +02:00 AGit
92 changed files with 588 additions and 324 deletions

View file

@ -1,6 +1,14 @@
{ {
"typeCheckingMode": "basic", "typeCheckingMode": "strict",
"pythonPlatform": "Linux", "pythonPlatform": "Linux",
"useLibraryCodeForTypes": true, "useLibraryCodeForTypes": true,
"verboseOutput": false "verboseOutput": false,
"reportImplicitOverride": "error",
"reportDeprecated": "error",
"reportMatchNotExhaustive": "error",
"reportUnnecessaryTypeIgnoreComment": "error",
"reportImplicitStringConcatenation": "error",
"reportUninitializedInstanceVariable": "error"
} }

View file

@ -2,6 +2,19 @@
{mypypath} {mypypath}
strict = true
warn_unreachable = true
enable_error_code = [
"explicit-override",
"ignore-without-code",
"possibly-undefined",
"truthy-bool",
"truthy-iterable",
"exhaustive-match",
]
[tool.isort] [tool.isort]
lines_between_sections = 1 lines_between_sections = 1

View file

@ -10,7 +10,7 @@ import sys
from enum import Enum, auto from enum import Enum, auto
from functools import cache from functools import cache
from typing import TYPE_CHECKING from typing import Any, cast, override, TYPE_CHECKING
from .lib.App import App as Base from .lib.App import App as Base
from .lib.Distro import Distro from .lib.Distro import Distro
@ -19,6 +19,7 @@ from .lib.ProjectConf import ProjectConf
if TYPE_CHECKING: if TYPE_CHECKING:
import argparse import argparse
from argparse import ArgumentParser
from typing import TypeAlias from typing import TypeAlias
@ -36,10 +37,10 @@ if TYPE_CHECKING:
class ResultCache(object): class ResultCache(object):
def __init__(self): def __init__(self) -> None:
self.__cache = {} self.__cache: dict[str, Any] = {}
def run(self, func, args): def run(self, func: Any, args: list[Any]) -> object:
d = self.__cache d = self.__cache
depth = 0 depth = 0
keys = [func.__name__] + args keys = [func.__name__] + args
@ -85,7 +86,7 @@ class App(Base):
return path return path
case 'relative': case 'relative':
return os.path.relpath(path) return os.path.relpath(path)
case None | 'absolute': case 'absolute':
return os.path.abspath(path) return os.path.abspath(path)
case _: case _:
m = re.search(r'^make:(\S+)$', fmt) m = re.search(r'^make:(\S+)$', fmt)
@ -136,21 +137,21 @@ class App(Base):
if search_absdirs is None: if search_absdirs is None:
search_absdirs = [] search_absdirs = []
def __format_relpath(path: str): def __format_relpath(path: str) -> str:
if path.startswith('./'): if path.startswith('./'):
return path[2:] return path[2:]
if path.endswith('/.'): if path.endswith('/.'):
return path[:-2] return path[:-2]
return path return path
def __relpath(target: str, base: str): def __relpath(target: str, base: str) -> str:
return __format_relpath(os.path.relpath(target, base)) return __format_relpath(os.path.relpath(target, base))
def __format_pd(name: str, pd: str, pretty: bool): def __format_pd(name: str, pd: str, pretty: bool) -> str | None:
if not pretty: if not pretty:
return pd return pd
if self.__topdir_fmt == 'absolute': if self.__topdir_fmt == 'absolute':
return os.path.abspath(pd) return str(os.path.abspath(pd))
if self.__topdir_fmt == 'unaltered': if self.__topdir_fmt == 'unaltered':
return pd return pd
if self.__topdir_fmt == 'relative': if self.__topdir_fmt == 'relative':
@ -174,6 +175,7 @@ class App(Base):
path = pd + '/' + sd path = pd + '/' + sd
if os.path.isdir(path): if os.path.isdir(path):
ret = __format_pd(name, pd, pretty) ret = __format_pd(name, pd, pretty)
assert ret is not None
if sd and sd[0] != '/': if sd and sd[0] != '/':
if ret == '.': if ret == '.':
ret = '' ret = ''
@ -200,11 +202,22 @@ class App(Base):
return None return None
def __get_project_refs_cached( def __get_project_refs_cached(
self, buf, visited, spec, section, key, add_self, scope, names_only self,
): buf: list[str],
return self.__res_cache.run( visited: set[str],
self.__get_project_refs, spec: str,
[buf, visited, spec, section, key, add_self, scope, names_only], section: str,
key: str,
add_self: bool,
scope: Scope,
names_only: bool,
) -> list[str]:
return cast(
'list[str]',
self.__res_cache.run(
self.__get_project_refs,
[buf, visited, spec, section, key, add_self, scope, names_only],
),
) )
def __get_project_refs( def __get_project_refs(
@ -283,7 +296,7 @@ class App(Base):
for dep in deps: for dep in deps:
self.__read_dep_graph([dep], sections, graph) self.__read_dep_graph([dep], sections, graph)
def __flip_dep_graph(self, graph: Graph): def __flip_dep_graph(self, graph: Graph) -> Graph:
ret: Graph = {} ret: Graph = {}
for project, deps in graph.items(): for project, deps in graph.items():
for d in deps: for d in deps:
@ -374,12 +387,14 @@ class App(Base):
default_pkg_filter = pkg_filter, default_pkg_filter = pkg_filter,
) )
async def __aexit__(self, exc_type, exc, tb) -> None: @override
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
if self.__exec_context is not None: if self.__exec_context is not None:
await self.__exec_context.close() await self.__exec_context.close()
self.__exec_context = None self.__exec_context = None
def _add_arguments(self, parser) -> None: @override
def _add_arguments(self, parser: ArgumentParser) -> None:
super()._add_arguments(parser) super()._add_arguments(parser)
parser.add_argument('-t', '--topdir', default = None, help = 'Project Path') parser.add_argument('-t', '--topdir', default = None, help = 'Project Path')
parser.add_argument( parser.add_argument(
@ -421,6 +436,7 @@ class App(Base):
help = 'Default filter for all distribution package-related operations', help = 'Default filter for all distribution package-related operations',
) )
@override
async def _run(self, args: argparse.Namespace) -> None: async def _run(self, args: argparse.Namespace) -> None:
self.___topdir = args.topdir self.___topdir = args.topdir
self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format) self.___pretty_topdir = self.__format_topdir(self.___topdir, args.topdir_format)
@ -482,7 +498,7 @@ class App(Base):
return self.__exec_context return self.__exec_context
@property @property
def top_name(self): def top_name(self) -> str | None:
return self.__top_name return self.__top_name
@property @property
@ -534,7 +550,7 @@ class App(Base):
def tmpl_dir(self, name: str) -> str | None: def tmpl_dir(self, name: str) -> str | None:
return self.find_dir(name, ['/tmpl'], ['/opt/' + name + '/share/tmpl']) return self.find_dir(name, ['/tmpl'], ['/opt/' + name + '/share/tmpl'])
def strip_module_from_spec(self, mod): def strip_module_from_spec(self, mod: str) -> str:
return re.sub(r'-dev$|-devel$|-run$', '', re.split('([=><]+)', mod)[0].strip()) return re.sub(r'-dev$|-devel$|-run$', '', re.split('([=><]+)', mod)[0].strip())
@cache @cache
@ -569,7 +585,7 @@ class App(Base):
return ret return ret
@cache @cache
def get_version(self, project) -> str: def get_version(self, project: str) -> str:
ret = self.get_value(project, 'version', '') ret = self.get_value(project, 'version', '')
if ret is None: if ret is None:
raise Exception(f"Can't get version of project {project}") raise Exception(f"Can't get version of project {project}")
@ -598,9 +614,9 @@ class App(Base):
projects: list[str], projects: list[str],
sections: list[str], sections: list[str],
keys: str | list[str], keys: str | list[str],
add_self: bool, scope: Scope = Scope.One,
scope: Scope, add_self: bool = False,
names_only = True, names_only: bool = False,
) -> list[str]: ) -> list[str]:
if isinstance(keys, str): if isinstance(keys, str):
keys = [keys] keys = [keys]

View file

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

View file

@ -1,11 +1,12 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import override, TYPE_CHECKING
from ..App import App as Parent from ..App import App as Parent
from ..CmdBase import CmdBase as Base from ..CmdBase import CmdBase as Base
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import Namespace
from typing import Iterable from typing import Iterable
from ..lib.Distro import Distro from ..lib.Distro import Distro
@ -20,7 +21,8 @@ class Cmd(Base): # export
) -> None: ) -> None:
super().__init__(parent, name, help, aliases = aliases) super().__init__(parent, name, help, aliases = aliases)
async def _run(self, args): @override
async def _run(self, args: 'Namespace') -> None:
# Missing subcommand # Missing subcommand
self.print_help(1) self.print_help(1)

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdPkg(Cmd): # export class CmdPkg(Cmd): # export
@ -18,12 +18,14 @@ class CmdPkg(Cmd): # export
) )
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdPlatform(Cmd): # export class CmdPlatform(Cmd): # export
@ -15,12 +15,14 @@ class CmdPlatform(Cmd): # export
) )
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdPosix(Cmd): # export class CmdPosix(Cmd): # export
@ -20,12 +20,14 @@ class CmdPosix(Cmd): # export
) )
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdProjects(Cmd): # export class CmdProjects(Cmd): # export
@ -17,12 +17,14 @@ class CmdProjects(Cmd): # export
) )
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdSecrets(Cmd): # export class CmdSecrets(Cmd): # export
@ -13,12 +13,14 @@ class CmdSecrets(Cmd): # export
super().__init__(parent, 'secrets', help = 'Manage package secrets') super().__init__(parent, 'secrets', help = 'Manage package secrets')
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

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

View file

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

View file

@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import override, TYPE_CHECKING
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
@ -16,6 +16,7 @@ class CmdInstall(Cmd): # export
help = "Install the distribution's notion of available packages", help = "Install the distribution's notion of available packages",
) )
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument('names', nargs = '*', help = 'Packages to be installed') 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>", help = "Don't expand macros in <names>",
) )
@override
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
names = ( names = (
args.names if args.fixed_strings else self.distro.expand_macros(args.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 __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .NamedPkgsCmd import NamedPkgsCmd from .NamedPkgsCmd import NamedPkgsCmd
@ -14,9 +14,11 @@ class CmdLs(NamedPkgsCmd): # export
def __init__(self, parent: Parent) -> None: def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'ls', help = 'List package contents') super().__init__(parent, 'ls', help = 'List package contents')
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
@override
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
for name in args.names: for name in args.names:
print('\n'.join(await self.distro.pkg_files(name))) print('\n'.join(await self.distro.pkg_files(name)))

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -1,11 +1,11 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdTar(Cmd): # export class CmdTar(Cmd): # export
@ -13,12 +13,14 @@ class CmdTar(Cmd): # export
super().__init__(parent, 'tar', help = 'Handle tar archives') super().__init__(parent, 'tar', help = 'Handle tar archives')
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
from contextlib import asynccontextmanager from contextlib import asynccontextmanager
from typing import TYPE_CHECKING from typing import Any, override, TYPE_CHECKING
from ....CmdBase import CmdBase as Base from ....CmdBase import CmdBase as Base
from ....lib.FileContext import FileContext from ....lib.FileContext import FileContext
@ -20,13 +20,14 @@ class Cmd(Base): # export
self.__tar_io: None = None self.__tar_io: None = None
@asynccontextmanager @asynccontextmanager
async def ctx(self, **kwargs) -> AsyncIterator[TarIo]: async def ctx(self, **kwargs: Any) -> AsyncIterator[TarIo]:
async with TarIo.create(src = self.app.args.archive_path, **kwargs) as ret: async with TarIo.create(src = self.app.args.archive_path, **kwargs) as ret:
ret.src.add_proc_filter( ret.src.add_proc_filter(
FileContext.Direction.In, ProcFilterGpg(ec = self.app.exec_context) FileContext.Direction.In, ProcFilterGpg(ec = self.app.exec_context)
) )
yield ret yield ret
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument( parser.add_argument(

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -1,10 +1,10 @@
from __future__ import annotations from __future__ import annotations
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
class CmdCheck(Cmd): # export class CmdCheck(Cmd): # export
@ -16,12 +16,14 @@ class CmdCheck(Cmd): # export
) )
self.load_subcommands() self.load_subcommands()
async def _run(self, args): @override
async def _run(self, args: Namespace) -> None:
import sys import sys
# Missing subcommand # Missing subcommand
self.parser.print_help() self.parser.print_help()
sys.exit(1) sys.exit(1)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)

View file

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

View file

@ -1,12 +1,13 @@
from argparse import ArgumentParser, ArgumentTypeError, Namespace from argparse import ArgumentParser, ArgumentTypeError, Namespace
from enum import Enum, auto from enum import Enum, auto
from typing import override
from ...lib.log import WARNING, log from ...lib.log import WARNING, log
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
from .lib.pkg_relations import VersionSyntax, pkg_relations from .lib.pkg_relations import VersionSyntax, pkg_relations
from .lib.templates import ListDict, RenderValues, tmpl_render from .lib.templates import ListDict, RenderValues, tmpl_render
def key_value(s): def key_value(s: str) -> tuple[str, str]:
try: try:
key, value = s.split('=', 1) key, value = s.split('=', 1)
except ValueError: except ValueError:
@ -50,8 +51,8 @@ class CmdCreateFile(Cmd): # export
self, self,
template_name: str, template_name: str,
values: list[RenderValues], values: list[RenderValues],
li_quote = False, li_quote: bool = False,
li_delimiter = '\n', li_delimiter: str = '\n',
) -> str: ) -> str:
return tmpl_render( return tmpl_render(
template_name, template_name,
@ -93,6 +94,7 @@ class CmdCreateFile(Cmd): # export
parent, 'create-file', help = 'Generate a file from project metadata' parent, 'create-file', help = 'Generate a file from project metadata'
) )
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument( parser.add_argument(
@ -124,6 +126,7 @@ class CmdCreateFile(Cmd): # export
) )
parser.add_argument('module', help = 'The module to generate the file for') parser.add_argument('module', help = 'The module to generate the file for')
@override
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
method = getattr(self, 'render_' + args.format, None) method = getattr(self, 'render_' + args.format, None)
if method is None: # Should be prevented by choices=[] but keeps linter happy if method is None: # Should be prevented by choices=[] but keeps linter happy

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -1,7 +1,7 @@
from __future__ import annotations from __future__ import annotations
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace from argparse import ArgumentParser, Namespace
@ -11,9 +11,11 @@ class CmdLibname(Cmd): # export
def __init__(self, parent: Parent) -> None: def __init__(self, parent: Parent) -> None:
super().__init__(parent, 'libname', help = 'libname') super().__init__(parent, 'libname', help = 'libname')
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument('module', nargs = '*', help = 'Modules') parser.add_argument('module', nargs = '*', help = 'Modules')
@override
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
print(self.app.get_libname(args.module)) 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.Uri import Uri
from ...lib.util import get_password, get_username, run_curl_into from ...lib.util import get_password, get_username, run_curl_into
from .Cmd import Cmd, Parent from .Cmd import Cmd, Parent
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser, Namespace from argparse import ArgumentParser, Namespace
@ -18,6 +18,7 @@ class CmdListRepos(Cmd): # export
parent, 'list-repos', help = 'Query a remote GIT server for repositories' parent, 'list-repos', help = 'Query a remote GIT server for repositories'
) )
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument( parser.add_argument(
@ -43,6 +44,7 @@ class CmdListRepos(Cmd): # export
'--from-owner', help = "List from-owner's projects", default = 'janware' '--from-owner', help = "List from-owner's projects", default = 'janware'
) )
@override
async def _run(self, args: Namespace) -> None: async def _run(self, args: Namespace) -> None:
base_url = Uri(args.base_url) base_url = Uri(args.base_url)
@ -141,6 +143,8 @@ class CmdListRepos(Cmd): # export
f'instance at {args.base_url}, tried {", ".join(tried)}' f'instance at {args.base_url}, tried {", ".join(tried)}'
) )
return return
case _:
pass
if os.path.isdir(args.base_url): if os.path.isdir(args.base_url):
for subdir in ['.', args.from_owner]: for subdir in ['.', args.from_owner]:
out = [] out = []

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -26,8 +26,8 @@ def pkg_relations(
ignore: set[str] = set(), ignore: set[str] = set(),
quote: bool = False, quote: bool = False,
skip_excluded: bool = False, skip_excluded: bool = False,
hide_self = False, hide_self: bool = False,
hide_jw_pkg = False, hide_jw_pkg: bool = False,
) -> list[str]: ) -> list[str]:
if subsections is None: if subsections is None:
@ -150,6 +150,8 @@ def pkg_relations(
match syntax: match syntax:
case VersionSyntax.semver: case VersionSyntax.semver:
pass pass
case VersionSyntax.names_only:
pass
case VersionSyntax.debian: case VersionSyntax.debian:
if len(expanded_dep) == 3: if len(expanded_dep) == 3:
match expanded_dep[1]: match expanded_dep[1]:
@ -157,11 +159,8 @@ def pkg_relations(
expanded_dep[1] = '<<' expanded_dep[1] = '<<'
case '>': case '>':
expanded_dep[1] = '>>' expanded_dep[1] = '>>'
case '_': case _:
raise NotImplementedError( pass
f'Unknown dependency syntax "{syntax}" for '
f'dependency "{dep[0]} {dep[1]} {dep[3]}"'
)
dep_str = ' '.join(expanded_dep) dep_str = ' '.join(expanded_dep)
if quote: if quote:
dep_str = '"' + dep_str + '"' dep_str = '"' + dep_str + '"'

View file

@ -1,5 +1,5 @@
import textwrap import textwrap
from typing import Iterable, TypeAlias, TypeGuard from typing import Any, Iterable, TypeAlias, TypeGuard
TupleList: TypeAlias = Iterable[tuple[str, str]] TupleList: TypeAlias = Iterable[tuple[str, str]]
ListDict: TypeAlias = dict[str, list[str]] ListDict: TypeAlias = dict[str, list[str]]
@ -66,17 +66,14 @@ def merge_values(*values: RenderValues) -> ListDict:
rhs_dict = render_values_to_list_dict(rhs) rhs_dict = render_values_to_list_dict(rhs)
for key, val in rhs_dict.items(): for key, val in rhs_dict.items():
entry = ret.setdefault(key, []) entry = ret.setdefault(key, [])
if isinstance(val, list): entry += val
entry += val
else:
entry.append(val)
return ret return ret
def format_list_dict( def format_list_dict(
template: str, values: ListDict | dict[str, str], li_quote: bool, li_delimiter: str template: str, values: ListDict | dict[str, str], li_quote: bool, li_delimiter: str
) -> str: ) -> str:
def __format_value(val): def __format_value(val: Any) -> str:
if not li_quote: if not li_quote:
return str(val) return str(val)
return f'"{val}"' return f'"{val}"'
@ -147,9 +144,9 @@ _templates = {
def tmpl_render( def tmpl_render(
template_name: str, template_name: str,
values: list[RenderValues], values: list[RenderValues],
li_quote = False, li_quote: bool = False,
li_delimiter = '\n', li_delimiter: str = '\n',
search_path: list[str] = [] search_path: list[str] | None = None,
) -> str: ) -> str:
def __format(template: str) -> str: def __format(template: str) -> str:
@ -160,7 +157,7 @@ def tmpl_render(
li_delimiter = li_delimiter, li_delimiter = li_delimiter,
) )
for d in search_path: for d in search_path if search_path else []:
path = d + '/' + template_name path = d + '/' + template_name
try: try:
with open(path, 'r') as f: with open(path, 'r') as f:

View file

@ -1,15 +1,15 @@
from __future__ import annotations from __future__ import annotations
from functools import cached_property from functools import cached_property
from typing import TYPE_CHECKING from typing import override, TYPE_CHECKING
from ...CmdBase import CmdBase as Base from ...CmdBase import CmdBase as Base
from ..CmdSecrets import CmdSecrets as Parent from ..CmdSecrets import CmdSecrets as Parent
from .lib.DistroContext import DistroContext from .lib.DistroContext import DistroContext
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Collection
from argparse import ArgumentParser from argparse import ArgumentParser
from typing import Iterable
from .lib.base import Attrs from .lib.base import Attrs
@ -19,33 +19,34 @@ class Cmd(Base): # export
def ctx(self) -> DistroContext: def ctx(self) -> DistroContext:
return DistroContext(self.app.distro) return DistroContext(self.app.distro)
async def _match_files(self, packages: Iterable[str], pattern: str) -> list[str]: async def _match_files(self, packages: Collection[str], pattern: str) -> list[str]:
return await self.ctx.match_files(packages, pattern) return await self.ctx.match_files(packages, pattern)
async def _list_template_files(self, packages: Iterable[str]) -> list[str]: async def _list_template_files(self, packages: Collection[str]) -> list[str]:
return await self.ctx.list_template_files(packages) return await self.ctx.list_template_files(packages)
async def _list_secret_paths( async def _list_secret_paths(
self, packages: Iterable[str], ignore_missing: bool = False self, packages: Collection[str], ignore_missing: bool = False
) -> list[str]: ) -> list[str]:
return await self.ctx.list_secret_paths(packages, ignore_missing) return await self.ctx.list_secret_paths(packages, ignore_missing)
async def _list_compilation_targets( async def _list_compilation_targets(
self, packages: Iterable[str], ignore_missing: bool = False self, packages: Collection[str], ignore_missing: bool = False
) -> list[str]: ) -> list[str]:
return await self.ctx.list_compilation_targets(packages, ignore_missing) return await self.ctx.list_compilation_targets(packages, ignore_missing)
async def _remove_compilation_targets(self, packages: Iterable[str]) -> list[str]: async def _remove_compilation_targets(self, packages: Collection[str]) -> list[str]:
return await self.ctx.remove_compilation_targets(packages) return await self.ctx.remove_compilation_targets(packages)
async def _compile_template_files( async def _compile_template_files(
self, packages: Iterable[str], default_attrs: Attrs self, packages: Collection[str], default_attrs: Attrs
) -> list[str]: ) -> list[str]:
return await self.ctx.compile_template_files(packages, default_attrs) return await self.ctx.compile_template_files(packages, default_attrs)
def __init__(self, parent: Parent, name: str, help: str) -> None: def __init__(self, parent: Parent, name: str, help: str) -> None:
super().__init__(parent, name, help) super().__init__(parent, name, help)
@override
def add_arguments(self, parser: ArgumentParser) -> None: def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser) super().add_arguments(parser)
parser.add_argument('packages', nargs = '*', help = 'Package names') parser.add_argument('packages', nargs = '*', help = 'Package names')

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -11,8 +11,8 @@ from ....lib.ProcFilterGpg import ProcFilterGpg
from .FilesContext import FilesContext from .FilesContext import FilesContext
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Collection
from .base import Attrs from .base import Attrs
from typing import Iterable
from ....lib.Distro import Distro from ....lib.Distro import Distro
@ -22,7 +22,7 @@ class DistroContext(FilesContext):
super().__init__(distro.ctx) super().__init__(distro.ctx)
self.__distro = distro self.__distro = distro
async def match_files(self, pkg_names: Iterable[str], pattern: str) -> list[str]: async def match_files(self, pkg_names: Collection[str], pattern: str) -> list[str]:
ret: list[str] = [] ret: list[str] = []
for pkg_name in pkg_names: for pkg_name in pkg_names:
for path in await self.__distro.pkg_files(pkg_name): for path in await self.__distro.pkg_files(pkg_name):
@ -30,13 +30,13 @@ class DistroContext(FilesContext):
ret.append(path) ret.append(path)
return ret return ret
async def list_template_files(self, pkg_names: Iterable[str]) -> list[str]: async def list_template_files(self, pkg_names: Collection[str]) -> list[str]:
if not pkg_names: if not pkg_names:
pkg_names = [p.name for p in await self.__distro.select()] pkg_names = [p.name for p in await self.__distro.select()]
return await self.match_files(pkg_names, pattern = r'.*\.jw-tmpl$') return await self.match_files(pkg_names, pattern = r'.*\.jw-tmpl$')
async def list_secret_paths( async def list_secret_paths(
self, pkg_names: Iterable[str], ignore_missing: bool = False self, pkg_names: Collection[str], ignore_missing: bool = False
) -> list[str]: ) -> list[str]:
ret = [] ret = []
for tmpl in await self.list_template_files(pkg_names): for tmpl in await self.list_template_files(pkg_names):
@ -47,7 +47,7 @@ class DistroContext(FilesContext):
return ret return ret
async def list_compilation_targets( async def list_compilation_targets(
self, pkg_names: Iterable[str], ignore_missing: bool = False self, pkg_names: Collection[str], ignore_missing: bool = False
) -> list[str]: ) -> list[str]:
ret = [] ret = []
for tmpl in await self.list_template_files(pkg_names): for tmpl in await self.list_template_files(pkg_names):
@ -57,7 +57,7 @@ class DistroContext(FilesContext):
ret.append(path) ret.append(path)
return ret return ret
async def remove_compilation_targets(self, pkg_names: Iterable[str]) -> list[str]: async def remove_compilation_targets(self, pkg_names: Collection[str]) -> list[str]:
ret: list[str] = [] ret: list[str] = []
for path in await self.list_compilation_targets(pkg_names): for path in await self.list_compilation_targets(pkg_names):
try: try:
@ -71,7 +71,7 @@ class DistroContext(FilesContext):
return ret return ret
async def compile_template_files( async def compile_template_files(
self, pkg_names: Iterable[str], default_attrs: Attrs self, pkg_names: Collection[str], default_attrs: Attrs
) -> list[str]: ) -> list[str]:
ret: list[str] = [] ret: list[str] = []
missing = 0 missing = 0
@ -96,7 +96,7 @@ class DistroContext(FilesContext):
async def install( async def install(
self, self,
src_uri: str, src_uri: str,
pkg_names: Iterable[str], pkg_names: Collection[str],
only_missing: bool = False, only_missing: bool = False,
verbose: bool = False, verbose: bool = False,
) -> None: ) -> None:

View file

@ -23,7 +23,11 @@ class FilesContext:
def ctx(self) -> FileContext: def ctx(self) -> FileContext:
return self.__ctx return self.__ctx
async def _read_key_value_file(self, path: str, throw = False) -> dict[str, str]: async def _read_key_value_file(
self,
path: str,
throw: bool = False,
) -> dict[str, str]:
ret: dict[str, str] = {} ret: dict[str, str] = {}
try: try:
result = await self.ctx.get(path) result = await self.ctx.get(path)

View file

@ -6,7 +6,7 @@ import os
import sys import sys
from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any from typing import Any, cast, override, TYPE_CHECKING
from .AsyncRunner import AsyncRunner from .AsyncRunner import AsyncRunner
from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level from .log import DEBUG, ERR, NOTICE, log, log_m, set_log_flags, set_log_level
@ -14,14 +14,14 @@ from .Types import LoadTypes
from .util import pretty_cmd from .util import pretty_cmd
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Awaitable from collections.abc import Awaitable, Collection
from typing import TypeVar from typing import TypeVar
T = TypeVar('T') T = TypeVar('T')
class App: # export class App: # export
def _add_arguments(self, parser): def _add_arguments(self, parser: ArgumentParser) -> None:
self.__parser.add_argument( self.__parser.add_argument(
'--log-flags', help = 'Log flags', default = self.__default_log_flags '--log-flags', help = 'Log flags', default = self.__default_log_flags
) )
@ -51,13 +51,16 @@ class App: # export
eloop: None = None, eloop: None = None,
) -> None: ) -> None:
def add_cmd_to_parser(cmd, parsers): def add_cmd_to_parser(cmd: AbstractCmd, parsers: Any) -> ArgumentParser:
parser = parsers.add_parser( parser = cast(
cmd.name, 'ArgumentParser',
help = cmd.help, parsers.add_parser(
description = cmd.description, cmd.name,
aliases = cmd.aliases, help = cmd.help,
formatter_class = ArgumentDefaultsHelpFormatter, description = cmd.description,
aliases = cmd.aliases,
formatter_class = ArgumentDefaultsHelpFormatter,
)
) )
parser.set_defaults(func = cmd.run) parser.set_defaults(func = cmd.run)
cmd.add_arguments(parser) cmd.add_arguments(parser)
@ -67,8 +70,8 @@ class App: # export
def add_cmds_to_parser( def add_cmds_to_parser(
parent: AbstractCmd | App, parent: AbstractCmd | App,
parser: ArgumentParser, parser: ArgumentParser,
cmds, cmds: Collection[AbstractCmd],
all = False all: bool = False
) -> None: ) -> None:
if not cmds: if not cmds:
return return
@ -150,7 +153,7 @@ class App: # export
cmd_classes: LoadTypes[AbstractCmd] = LoadTypes( cmd_classes: LoadTypes[AbstractCmd] = LoadTypes(
modules if modules else ['__main__'], modules if modules else ['__main__'],
type_name_filter = name_filter, type_name_filter = name_filter,
type_filter = [AbstractCmd], # type: ignore[type-abstract] type_filter = [AbstractCmd],
) )
add_all_parsers = ( add_all_parsers = (
'-h' in sys.argv or '--help' in sys.argv or '_ARGCOMPLETE' in os.environ '-h' in sys.argv or '--help' in sys.argv or '_ARGCOMPLETE' in os.environ
@ -168,7 +171,7 @@ class App: # export
'-h', '--help', action = 'help', help = 'Show this help message and exit' '-h', '--help', action = 'help', help = 'Show this help message and exit'
) )
def __del__(self): def __del__(self) -> None:
if self.__own_eloop: if self.__own_eloop:
if self.__eloop is not None: if self.__eloop is not None:
self.__eloop.close() self.__eloop.close()
@ -178,24 +181,27 @@ class App: # export
async def __aenter__(self) -> None: async def __aenter__(self) -> None:
pass pass
async def __aexit__(self, exc_type, exc, tb) -> None: async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
pass pass
async def __run(self, argv = None) -> None: async def __run(self, argv: list[str] | None = None) -> None:
try: try:
# Import argcomplete only here to not require it to be compatible # Import argcomplete only here to not require it to be compatible
# with minimal environments # with minimal environments
from argcomplete.completers import ( # type: ignore[import-not-found] from argcomplete.completers import ( # type: ignore[import-not-found, unused-ignore]
BaseCompleter BaseCompleter
) )
class NoopCompleter(BaseCompleter): class NoopCompleter(BaseCompleter): # type: ignore[misc, unused-ignore]
def __call__(self, *args, **kwargs): @override
def __call__( # pyright: ignore[reportGeneralTypeIssues]
self, *args: Any, **kwargs: Any
) -> None:
return None return None
import argcomplete # type: ignore[import-not-found] import argcomplete # type: ignore[import-not-found, unused-ignore]
argcomplete.autocomplete(self.__parser, default_completer = NoopCompleter()) argcomplete.autocomplete(self.__parser, default_completer = NoopCompleter())
@ -246,7 +252,7 @@ class App: # export
self.__parser.print_help() self.__parser.print_help()
return None return None
# Run sub-command. Overwrite if you want to do anything before or after # Run sub-command. Overwrite if you want to do anything before or after
return await self.args.func(args) return cast('None | int', await self.args.func(args))
def call_async(self, awaitable: Awaitable[T], timeout: float | None = None) -> T: def call_async(self, awaitable: Awaitable[T], timeout: float | None = None) -> T:
return self.async_runner.call(awaitable, timeout) return self.async_runner.call(awaitable, timeout)
@ -283,7 +289,7 @@ class App: # export
def parser(self) -> ArgumentParser: def parser(self) -> ArgumentParser:
return self.__parser return self.__parser
def run(self, argv = None) -> None: def run(self, argv: list[str] | None = None) -> None:
try: try:
ret = self.eloop.run_until_complete(self.__run(argv)) ret = self.eloop.run_until_complete(self.__run(argv))
finally: finally:
@ -293,7 +299,10 @@ class App: # export
return ret return ret
def run_sub_commands( # export def run_sub_commands( # export
description = '', name_filter = '^Cmd.*', modules = None, argv = None description: str = '',
): name_filter: str = '^Cmd.*',
modules: list[str] | None = None,
argv: list[str] | None = None,
) -> None:
app = App(description, name_filter, modules) app = App(description, name_filter, modules)
return app.run(argv = argv) app.run(argv = argv)

View file

@ -4,7 +4,7 @@ import asyncio
import concurrent.futures import concurrent.futures
import contextlib import contextlib
from typing import TypeVar, TYPE_CHECKING from typing import Any, TypeVar, TYPE_CHECKING, cast
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Awaitable, Generator from collections.abc import Awaitable, Generator
@ -44,8 +44,11 @@ class AsyncRunner:
self._loop = self._cm.__enter__() self._loop = self._cm.__enter__()
def call(self, awaitable: Awaitable[T], timeout: float | None = None) -> T: def call(self, awaitable: Awaitable[T], timeout: float | None = None) -> T:
fut = asyncio.run_coroutine_threadsafe(awaitable, self._loop) # type: ignore fut = asyncio.run_coroutine_threadsafe(
return fut.result(timeout) awaitable, # type: ignore[arg-type, var-annotated]
self._loop,
)
return cast('T', fut.result(timeout))
def close(self) -> None: def close(self) -> None:
self._cm.__exit__(None, None, None) self._cm.__exit__(None, None, None)
@ -53,5 +56,5 @@ class AsyncRunner:
def __enter__(self) -> AsyncRunner: def __enter__(self) -> AsyncRunner:
return self return self
def __exit__(self, exc_type, exc, tb) -> None: def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
self.close() self.close()

View file

@ -3,14 +3,14 @@ from __future__ import annotations
import abc import abc
import sys import sys
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any, override
from .log import ERR from .log import ERR
from .Types import LoadTypes, Types from .Types import LoadTypes, Types
if TYPE_CHECKING: if TYPE_CHECKING:
from argparse import ArgumentParser from argparse import ArgumentParser, Namespace
from typing import Any, Iterable from typing import Iterable
from .App import App from .App import App
@ -26,7 +26,7 @@ class AbstractCmd(abc.ABC):
self.__child_classes: list[type[Cmd]] = [] self.__child_classes: list[type[Cmd]] = []
self.__parser: ArgumentParser | None = None self.__parser: ArgumentParser | None = None
def set_parent(self, parent: Any | Cmd): def set_parent(self, parent: Any | Cmd) -> None:
self.__parent = parent self.__parent = parent
@property @property
@ -76,7 +76,7 @@ class AbstractCmd(abc.ABC):
return self.__parser return self.__parser
# Don't use a setter decorator to force using a grepable method # Don't use a setter decorator to force using a grepable method
def set_parser(self, parser: ArgumentParser): def set_parser(self, parser: ArgumentParser) -> None:
self.__parser = parser self.__parser = parser
def print_help(self, exit_status: int | None = None) -> None: def print_help(self, exit_status: int | None = None) -> None:
@ -84,10 +84,11 @@ class AbstractCmd(abc.ABC):
if exit_status is not None: if exit_status is not None:
sys.exit(exit_status) sys.exit(exit_status)
def add_subcommands(self, cmds: Cmd | list[Cmd] | Types | list[Types]) -> None: def add_subcommands(
self, cmds: Cmd | list[Cmd] | Types[Any] | list[Types[Any]]
) -> None:
if isinstance(cmds, Cmd): if isinstance(cmds, Cmd):
assert False raise NotImplementedError('Single Cmd should be handled elsewhere')
return
if isinstance(cmds, list): if isinstance(cmds, list):
for cmd in cmds: for cmd in cmds:
self.add_subcommands(cmd) self.add_subcommands(cmd)
@ -128,11 +129,11 @@ class AbstractCmd(abc.ABC):
pass pass
@abc.abstractmethod @abc.abstractmethod
async def _run(self, args) -> None: async def _run(self, args: Namespace) -> None:
if isinstance(self.__parent, Cmd): # Calling App.run() would loop if isinstance(self.__parent, Cmd): # Calling App.run() would loop
return await self.__parent._run(args) return await self.__parent._run(args)
async def run(self, args): async def run(self, args: Namespace) -> None:
return await self._run(args) return await self._run(args)
@abc.abstractmethod @abc.abstractmethod
@ -173,14 +174,18 @@ class Cmd(AbstractCmd): # export
self.__description = description if description else help self.__description = description if description else help
self.__aliases = aliases if aliases else [] self.__aliases = aliases if aliases else []
@override
def _name(self) -> str: def _name(self) -> str:
return self.__name return self.__name
@override
def _help(self) -> str: def _help(self) -> str:
return self.__help return self.__help
@override
def _description(self) -> str: def _description(self) -> str:
return self.__description return self.__description
@override
def _aliases(self) -> Iterable[str]: def _aliases(self) -> Iterable[str]:
return self.__aliases return self.__aliases

View file

@ -1,4 +1,4 @@
from typing import Self from typing import Any, Self
from .FileContext import FileContext from .FileContext import FileContext
from .Uri import Uri from .Uri import Uri
@ -9,12 +9,10 @@ class CopyContext:
self, self,
src: Uri | str | FileContext, src: Uri | str | FileContext,
dst: Uri | str | FileContext, dst: Uri | str | FileContext,
chroot = False chroot: bool = False,
) -> None: ) -> None:
def __uri(ctx: FileContext | Uri | str) -> Uri | str | None: def __uri(ctx: FileContext | Uri | str) -> Uri | str:
if ctx is None:
return None
if isinstance(ctx, Uri): if isinstance(ctx, Uri):
return ctx return ctx
if isinstance(ctx, str): if isinstance(ctx, str):
@ -45,7 +43,7 @@ class CopyContext:
await self.__dst.open() await self.__dst.open()
return self return self
async def __aexit__(self, exc_type, exc, tb): async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
if self.__src is not None: if self.__src is not None:
await self.__src.close() await self.__src.close()
self.__src = None self.__src = None

View file

@ -6,12 +6,13 @@ import re
import sys import sys
from functools import cached_property from functools import cached_property
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any
from .log import ERR, INFO, WARNING, log from .log import ERR, INFO, WARNING, log
from .base import InputMode from .base import InputMode
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Collection
from typing import Iterable from typing import Iterable
from .base import Result from .base import Result
@ -97,6 +98,8 @@ class Distro(abc.ABC):
match ret: match ret:
case 'opensuse-tumbleweed': case 'opensuse-tumbleweed':
return 'suse' return 'suse'
case _:
pass
return ret return ret
@classmethod @classmethod
@ -105,8 +108,8 @@ class Distro(abc.ABC):
ec: ExecContext, ec: ExecContext,
id: str | None = None, id: str | None = None,
os_release_str: str | None = None, os_release_str: str | None = None,
**kwargs, **kwargs: Any,
): ) -> 'Distro':
if id is None: if id is None:
os_release_str = await cls.read_os_release_str(ec) os_release_str = await cls.read_os_release_str(ec)
id = cls.parse_os_release_field_id(os_release_str) id = cls.parse_os_release_field_id(os_release_str)
@ -118,6 +121,8 @@ class Distro(abc.ABC):
backend_id = 'redhat' backend_id = 'redhat'
case 'opensuse' | 'suse': case 'opensuse' | 'suse':
backend_id = 'suse' backend_id = 'suse'
case _:
pass
module_path = 'jw.pkg.lib.distros.' + backend_id + '.Distro' module_path = 'jw.pkg.lib.distros.' + backend_id + '.Distro'
try: try:
module = importlib.import_module(module_path) module = importlib.import_module(module_path)
@ -137,8 +142,9 @@ class Distro(abc.ABC):
@cached_property @cached_property
def os_cascade(self) -> list[str]: def os_cascade(self) -> list[str]:
ret: list[str] = []
def __append(entry: str): def __append(entry: str) -> None:
if entry not in ret: if entry not in ret:
ret.append(entry) ret.append(entry)
@ -172,6 +178,8 @@ class Distro(abc.ABC):
__append('linux') __append('linux')
__append('pkg-pm') __append('pkg-pm')
__append('pm-pacman') __append('pm-pacman')
case _:
pass
os = self.os os = self.os
name = re.sub(r'-.*', '', os) name = re.sub(r'-.*', '', os)
@ -278,7 +286,7 @@ class Distro(abc.ABC):
def macros(cls) -> list[str]: def macros(cls) -> list[str]:
return ['%%{' + name + '}' for name in cls.macro_names] return ['%%{' + name + '}' for name in cls.macro_names]
def expand_macros(self, fmt: str | Iterable) -> str | list[str]: def expand_macros(self, fmt: str | Iterable[str]) -> str | list[str]:
ret: str | list[str] ret: str | list[str]
if not isinstance(fmt, str): if not isinstance(fmt, str):
ret = [] ret = []
@ -315,10 +323,10 @@ class Distro(abc.ABC):
def default_pkg_filter(self) -> PackageFilter | None: def default_pkg_filter(self) -> PackageFilter | None:
return self.__default_pkg_filter return self.__default_pkg_filter
async def run(self, *args, **kwargs) -> Result: async def run(self, *args: Any, **kwargs: Any) -> Result:
return await self.__exec_context.run(*args, **kwargs) return await self.__exec_context.run(*args, **kwargs)
async def sudo(self, *args, **kwargs) -> Result: async def sudo(self, *args: Any, **kwargs: Any) -> Result:
return await self.__exec_context.sudo(*args, **kwargs) return await self.__exec_context.sudo(*args, **kwargs)
@property @property
@ -359,17 +367,16 @@ class Distro(abc.ABC):
# -- select # -- select
@abc.abstractmethod @abc.abstractmethod
async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]: async def _select_by_name(self, names: Collection[str]) -> Iterable[Package]:
pass pass
async def _select(self, names: Iterable[str], async def _select(self, names: Collection[str],
filter: PackageFilter) -> Iterable[Package]: filter: PackageFilter) -> Iterable[Package]:
assert filter, 'No filter in _select()'
return [p for p in await self._select_by_name(names) if filter.match(p)] return [p for p in await self._select_by_name(names) if filter.match(p)]
async def select( async def select(
self, self,
names: Iterable[str] = [], names: Collection[str] = [],
filter: PackageFilter | None = None filter: PackageFilter | None = None
) -> Iterable[Package]: ) -> Iterable[Package]:
if not filter: if not filter:
@ -435,7 +442,7 @@ class Distro(abc.ABC):
if names: if names:
await self._install(names, only_update = only_update) await self._install(names, only_update = only_update)
async def install(self, names: Iterable[str], only_update: bool = False) -> None: async def install(self, names: Collection[str], only_update: bool = False) -> None:
if not names: if not names:
log(WARNING, 'No packages specified for installation') log(WARNING, 'No packages specified for installation')
return return
@ -447,7 +454,7 @@ class Distro(abc.ABC):
async def _delete(self, names: Iterable[str]) -> None: async def _delete(self, names: Iterable[str]) -> None:
pass pass
async def delete(self, names: Iterable[str]) -> None: async def delete(self, names: Collection[str]) -> None:
if not names: if not names:
log(WARNING, 'No packages specified for deletion') log(WARNING, 'No packages specified for deletion')
return return

View file

@ -5,7 +5,7 @@ import errno
import sys import sys
from decimal import ROUND_FLOOR, Decimal from decimal import ROUND_FLOOR, Decimal
from typing import TYPE_CHECKING, NamedTuple from typing import Any, override, TYPE_CHECKING, NamedTuple
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Type from typing import Type
@ -150,9 +150,7 @@ class ExecContext(Base):
interactive = sys.stdin.isatty() interactive = sys.stdin.isatty()
else: else:
interactive = False interactive = False
if cmd_input is None: if isinstance(cmd_input, str):
cmd_input_bytes = None
elif isinstance(cmd_input, str):
cmd_input_bytes = cmd_input.encode(sys.stdout.encoding or 'utf-8') cmd_input_bytes = cmd_input.encode(sys.stdout.encoding or 'utf-8')
else: else:
cmd_input_bytes = cmd_input cmd_input_bytes = cmd_input
@ -216,7 +214,7 @@ class ExecContext(Base):
self.__pretty_cmd = pretty_cmd(self.__cmd, self.__wd) self.__pretty_cmd = pretty_cmd(self.__cmd, self.__wd)
return self.__pretty_cmd return self.__pretty_cmd
def log(self, prio: int, *args, **kwargs) -> None: def log(self, prio: int, *args: Any, **kwargs: Any) -> None:
log(prio, self.__log_prefix, *args, **kwargs) log(prio, self.__log_prefix, *args, **kwargs)
def log_delim(self, start: bool) -> None: def log_delim(self, start: bool) -> None:
@ -246,11 +244,12 @@ class ExecContext(Base):
def __mode_str(cls, mode: int) -> str: def __mode_str(cls, mode: int) -> str:
return f'{mode:0o}' return f'{mode:0o}'
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@classmethod @classmethod
def create(cls, *args, **kwargs) -> ExecContext: @override
def create(cls, *args: Any, **kwargs: Any) -> ExecContext:
ret = super().create(*args, **kwargs) ret = super().create(*args, **kwargs)
if not isinstance(ret, cls): if not isinstance(ret, cls):
raise TypeError(f'Expected {cls.__name__}, got {type(ret).__name__}') raise TypeError(f'Expected {cls.__name__}, got {type(ret).__name__}')
@ -451,6 +450,7 @@ class ExecContext(Base):
cc.check_exit_code(ret) cc.check_exit_code(ret)
return ret return ret
@override
async def _get( async def _get(
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result: ) -> Result:
@ -485,6 +485,7 @@ class ExecContext(Base):
cc.check_exit_code(ret) cc.check_exit_code(ret)
return ret return ret
@override
async def _put( async def _put(
self, self,
path: str, path: str,
@ -504,7 +505,7 @@ class ExecContext(Base):
async def __run( async def __run(
cmd: list[str], cmd: list[str],
cmd_input: Input = InputMode.NonInteractive, cmd_input: Input = InputMode.NonInteractive,
**kwargs **kwargs: Any
) -> Result: ) -> Result:
return await self.run(cmd, cmd_input = cmd_input, **kwargs) return await self.run(cmd, cmd_input = cmd_input, **kwargs)
@ -565,22 +566,27 @@ class ExecContext(Base):
log(ERR, msg) log(ERR, msg)
return ret return ret
@override
async def _unlink(self, path: str) -> None: async def _unlink(self, path: str) -> None:
cmd = ['rm', '-f', path] cmd = ['rm', '-f', path]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _erase(self, path: str) -> None: async def _erase(self, path: str) -> None:
cmd = ['rm', '-rf', path] cmd = ['rm', '-rf', path]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _rename(self, src: str, dst: str) -> None: async def _rename(self, src: str, dst: str) -> None:
cmd = ['mv', src, dst] cmd = ['mv', src, dst]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _mkdir(self, path: str, mode: int) -> None: async def _mkdir(self, path: str, mode: int) -> None:
cmd = ['mkdir', path, '-m', self.__mode_str(mode)] cmd = ['mkdir', path, '-m', self.__mode_str(mode)]
await self.run(cmd, cmd_input = InputMode.NonInteractive) await self.run(cmd, cmd_input = InputMode.NonInteractive)
@override
async def _mktemp(self, tmpl: str, directory: bool) -> str: async def _mktemp(self, tmpl: str, directory: bool) -> str:
cmd = ['mktemp'] cmd = ['mktemp']
if directory: if directory:
@ -593,6 +599,7 @@ class ExecContext(Base):
) )
return result.stdout_str.strip() return result.stdout_str.strip()
@override
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult: async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
async def __stat(opts: list[str]) -> Result: async def __stat(opts: list[str]) -> Result:
@ -662,6 +669,7 @@ class ExecContext(Base):
assert stdout is not None # Just there to pacify the linter assert stdout is not None # Just there to pacify the linter
return _build_stat_result(stdout.rstrip('\n').split(_US), mode_base = 8) 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: async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
if owner is None and group is None: if owner is None and group is None:
raise ValueError(f'Tried to chown("{path}") without owner and group') raise ValueError(f'Tried to chown("{path}") without owner and group')
@ -674,6 +682,7 @@ class ExecContext(Base):
assert ownership is not None # Impossible, just there to calm the linter assert ownership is not None # Impossible, just there to calm the linter
await self.run(['chown', ownership, path], cmd_input = InputMode.NonInteractive) await self.run(['chown', ownership, path], cmd_input = InputMode.NonInteractive)
@override
async def _chmod(self, path: str, mode: int) -> None: async def _chmod(self, path: str, mode: int) -> None:
await self.run( await self.run(
['chmod', self.__mode_str(mode), path], ['chmod', self.__mode_str(mode), path],

View file

@ -4,7 +4,7 @@ import abc
from enum import Enum, auto from enum import Enum, auto
from functools import cached_property from functools import cached_property
from typing import TYPE_CHECKING from typing import Any, override, TYPE_CHECKING
from .log import DEBUG, ERR, log from .log import DEBUG, ERR, log
from .Uri import Uri from .Uri import Uri
@ -24,7 +24,7 @@ class FileContext(abc.ABC):
self, self,
uri: str | Uri, uri: str | Uri,
interactive: bool | None = None, interactive: bool | None = None,
verbose_default = False, verbose_default: bool = False,
chroot: bool = False, chroot: bool = False,
in_pipe: ProcPipeline | None = None, in_pipe: ProcPipeline | None = None,
out_pipe: ProcPipeline | None = None, out_pipe: ProcPipeline | None = None,
@ -43,17 +43,18 @@ class FileContext(abc.ABC):
f'= "{verbose_default}"' f'= "{verbose_default}"'
) )
async def __aenter__(self): async def __aenter__(self) -> 'FileContext':
await self.open() await self.open()
return self return self
async def __aexit__(self, exc_type, exc, tb): async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
await self.close() await self.close()
@override
def __repr__(self) -> str: def __repr__(self) -> str:
return self.__uri.id return self.__uri.id
def __pipe(self, d: Direction): def __pipe(self, d: Direction) -> 'ProcPipeline':
match d: match d:
case self.Direction.In: case self.Direction.In:
if not self.__in_pipe: if not self.__in_pipe:
@ -75,7 +76,7 @@ class FileContext(abc.ABC):
return self.root + path return self.root + path
return self.root + '/' + path return self.root + '/' + path
def add_proc_filter(self, d: Direction, proc_filter: ProcFilter): def add_proc_filter(self, d: Direction, proc_filter: ProcFilter) -> None:
self.__pipe(d).append(proc_filter) self.__pipe(d).append(proc_filter)
async def _open(self) -> None: async def _open(self) -> None:
@ -296,13 +297,12 @@ class FileContext(abc.ABC):
except Exception as e: except Exception as e:
log(ERR, f'{self.log_name}: Failed to stat({path}) ({str(e)})') log(ERR, f'{self.log_name}: Failed to stat({path}) ({str(e)})')
raise raise
return False
async def is_dir(self, path: str, follow_symlinks = True) -> bool: async def is_dir(self, path: str, follow_symlinks: bool = True) -> bool:
return await self._is_dir(self._chroot(path), follow_symlinks = follow_symlinks) return await self._is_dir(self._chroot(path), follow_symlinks = follow_symlinks)
@classmethod @classmethod
def create(cls, uri: str | Uri, *args, **kwargs) -> FileContext: def create(cls, uri: str | Uri, *args: Any, **kwargs: Any) -> 'FileContext':
uri = Uri.pimp(uri) uri = Uri.pimp(uri)
match uri.protocol: match uri.protocol:
case 'local' | 'file': case 'local' | 'file':

View file

@ -1,4 +1,6 @@
from typing import Any from __future__ import annotations
from typing import override, Any
meta_tags = [ meta_tags = [
'name', 'name',
@ -16,7 +18,7 @@ class Package:
maintainer: str | None = None maintainer: str | None = None
@classmethod @classmethod
def parse_spec_str(cls, spec: str, delimiter = '|'): def parse_spec_str(cls, spec: str, delimiter: str = '|') -> 'Package':
tags = spec.split(delimiter) tags = spec.split(delimiter)
if len(tags) != 5: if len(tags) != 5:
raise ValueError(f'Invalid package spec string "{spec}"') raise ValueError(f'Invalid package spec string "{spec}"')
@ -29,14 +31,14 @@ class Package:
) )
@classmethod @classmethod
def parse_specs_str(cls, specs: str, delimiter = '|'): def parse_specs_str(cls, specs: str, delimiter: str = '|') -> list[Package]:
ret: list[Package] = [] ret: list[Package] = []
for spec in specs.splitlines(): for spec in specs.splitlines():
ret.append(cls.parse_spec_str(spec)) ret.append(cls.parse_spec_str(spec))
return ret return ret
@classmethod @classmethod
def order_tags(cls, mapping: dict[str, Any]): def order_tags(cls, mapping: dict[str, Any]) -> dict[str, Any]:
ret: dict[str, Any] = {} ret: dict[str, Any] = {}
for tag in meta_tags: for tag in meta_tags:
ret[tag] = mapping.get(tag, '') ret[tag] = mapping.get(tag, '')
@ -56,6 +58,7 @@ class Package:
self.url = url self.url = url
self.maintainer = maintainer self.maintainer = maintainer
@override
def __repr__(self) -> str: def __repr__(self) -> str:
global meta_tags global meta_tags
return '\n'.join([f'{key:<15}: {getattr(self, key)}' for key in 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 abc
import re import re
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
if TYPE_CHECKING: if TYPE_CHECKING:
from .Package import Package from .Package import Package
@ -25,6 +25,7 @@ class PackageFilterString(PackageFilter):
raise Exception(f'Unsupported filter string "{definition}"') raise Exception(f'Unsupported filter string "{definition}"')
self.__definition = url_rx_str self.__definition = url_rx_str
@override
def _match(self, package: Package) -> bool: def _match(self, package: Package) -> bool:
url = package.url url = package.url
if url is None: if url is None:

View file

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

View file

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

View file

@ -1,5 +1,7 @@
from __future__ import annotations from __future__ import annotations
from typing import override
class Result: class Result:
def __init__( def __init__(
@ -31,7 +33,7 @@ class Result:
def __try_decode( def __try_decode(
self, self,
stdxxx: bytes | None, stdxxx: bytes | None,
quote = False, quote: bool = False,
truncate: int | None = None, truncate: int | None = None,
annotate: bool = True, annotate: bool = True,
label: str | None = None, label: str | None = None,
@ -57,7 +59,7 @@ class Result:
self, self,
cmd: list[str] | None = None, cmd: list[str] | None = None,
wd: str | None = None, wd: str | None = None,
verbose = True verbose: bool = True,
) -> str: ) -> str:
def __status_str(cmd: list[str] | None, wd: str | None, verbose: bool) -> str: def __status_str(cmd: list[str] | None, wd: str | None, verbose: bool) -> str:
@ -113,6 +115,7 @@ class Result:
return ret return ret
@override
def __repr__(self) -> str: def __repr__(self) -> str:
return self.__summarize(verbose = False) return self.__summarize(verbose = False)
@ -156,8 +159,6 @@ class Result:
if self.status == 0: if self.status == 0:
return False return False
err = self.stderr_str err = self.stderr_str
if err is None:
return False
import re import re
return re.search(pattern, err) is not None return re.search(pattern, err) is not None

View file

@ -9,14 +9,14 @@ from tarfile import TarFile, TarInfo
from .CopyContext import CopyContext from .CopyContext import CopyContext
from .ExecContext import ExecContext from .ExecContext import ExecContext
from .log import DEBUG, ERR, log from .log import DEBUG, ERR, log
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any, override
if TYPE_CHECKING: if TYPE_CHECKING:
from .base import StatResult from .base import StatResult
class TarIo(CopyContext): class TarIo(CopyContext):
def __init__(self, *args, **kwargs) -> None: def __init__(self, *args: Any, **kwargs: Any) -> None:
kwargs['chroot'] = False kwargs['chroot'] = False
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@ -44,7 +44,7 @@ class TarIo(CopyContext):
async def _read_filtered( async def _read_filtered(
self, self,
path, path: str,
path_filter: list[str] | None = None, path_filter: list[str] | None = None,
matched: list[str] | None = None, matched: list[str] | None = None,
) -> bytes: ) -> bytes:
@ -81,7 +81,7 @@ class TarIo(CopyContext):
return ret return ret
@classmethod @classmethod
def create(cls, *args, type: str | None = None, **kwargs): def create(cls, *args: Any, type: str | None = None, **kwargs: Any) -> 'TarIo':
if type is not None: if type is not None:
raise NotImplementedError raise NotImplementedError
# return TarIoTarFile(*args, **kwargs) # return TarIoTarFile(*args, **kwargs)
@ -89,6 +89,7 @@ class TarIo(CopyContext):
class TarIoTarFile(TarIo): class TarIoTarFile(TarIo):
@override
async def _extract(self, blob: bytes, root: str | None = None) -> None: async def _extract(self, blob: bytes, root: str | None = None) -> None:
tf = TarFile(fileobj = io.BytesIO(blob)) tf = TarFile(fileobj = io.BytesIO(blob))
for info in tf.getmembers(): for info in tf.getmembers():
@ -111,7 +112,16 @@ class TarIoTarFile(TarIo):
class TarIoTarExec(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 @property
@override
def dst(self) -> ExecContext: def dst(self) -> ExecContext:
ret = super().dst ret = super().dst
if not isinstance(ret, ExecContext): if not isinstance(ret, ExecContext):
@ -120,10 +130,3 @@ class TarIoTarExec(TarIo):
'context, which only has a file context' 'context, which only has a file context'
) )
return ret 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 re
import sys import sys
from typing import TYPE_CHECKING, Generic, Iterable, TypeVar from typing import Any, override, TYPE_CHECKING, Generic, Iterable, TypeVar
from .log import ERR, OFF, log, parse_log_level 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 class Types(abc.ABC, Iterable[type[T]], Generic[T]): # export
@override
def __iter__(self) -> Iterator[type[T]]: def __iter__(self) -> Iterator[type[T]]:
return iter(self._classes()) return iter(self._classes())
@ -32,7 +33,7 @@ class Types(abc.ABC, Iterable[type[T]], Generic[T]): # export
def _stringify(self) -> list[str]: def _stringify(self) -> list[str]:
pass pass
def dump(self, prio: int, *args, **kwargs) -> None: def dump(self, prio: int, *args: Any, **kwargs: Any) -> None:
contents = self._stringify() contents = self._stringify()
log(prio, ',--- ', *args, **kwargs) log(prio, ',--- ', *args, **kwargs)
for line in contents: for line in contents:
@ -46,8 +47,8 @@ class LoadTypes(Types[T]): # export
mod_names: Iterable[str], mod_names: Iterable[str],
type_name_filter: str | None = None, type_name_filter: str | None = None,
type_filter: Sequence[type[Any]] | None = None, type_filter: Sequence[type[Any]] | None = None,
debug_level = None, debug_level: int | None = None,
): ) -> None:
if debug_level is None: if debug_level is None:
val = os.getenv('JW_LOG_LEVEL_LOAD_TYPES') val = os.getenv('JW_LOG_LEVEL_LOAD_TYPES')
if val is not None: if val is not None:
@ -60,11 +61,12 @@ class LoadTypes(Types[T]): # export
self.__mod_names = mod_names self.__mod_names = mod_names
self.__classes: list[type[T]] | None = None self.__classes: list[type[T]] | None = None
def _debug(self, *args, **kwargs) -> None: def _debug(self, *args: Any, **kwargs: Any) -> None:
if self.__debug_level != OFF: if self.__debug_level != OFF:
log(self.__debug_level, *args, **kwargs) log(self.__debug_level, *args, **kwargs)
def _stringify(self): @override
def _stringify(self) -> list[str]:
tf = 'None' if self.__type_filter is None else ( tf = 'None' if self.__type_filter is None else (
', '.join([str(f) for f in self.__type_filter]) ', '.join([str(f) for f in self.__type_filter])
) )
@ -74,6 +76,7 @@ class LoadTypes(Types[T]): # export
'mod_names: ' + ', '.join(self.__mod_names), 'mod_names: ' + ', '.join(self.__mod_names),
] ]
@override
def _classes(self) -> Iterable[type[T]]: def _classes(self) -> Iterable[type[T]]:
if self.__classes is None: if self.__classes is None:

View file

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

View file

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

View file

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

View file

@ -1,11 +1,12 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any, override
from ...Distro import Distro as Base from ...Distro import Distro as Base
from ...pm.rpm import list_files, query_packages, run_rpm from ...pm.rpm import list_files, query_packages, run_rpm
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Collection
from typing import Iterable from typing import Iterable
from ...base import Result from ...base import Result
@ -34,18 +35,25 @@ class Distro(Base):
if sudo else await self.run(cmd, verbose = verbose) if sudo else await self.run(cmd, verbose = verbose)
) )
async def rpm(self, *args, ec: ExecContext | None = None, **kwargs) -> str: async def rpm(
self,
*args: Any,
ec: ExecContext | None = None,
**kwargs: Any,
) -> str:
if ec is None: if ec is None:
ec = self.ctx ec = self.ctx
kwargs['ec'] = ec kwargs['ec'] = ec
return await run_rpm(*args, **kwargs) return await run_rpm(*args, **kwargs)
def __init__(self, *args, **kwargs): def __init__(self, *args: Any, **kwargs: Any) -> None:
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@override
async def _ref(self) -> None: async def _ref(self) -> None:
await self.zypper(['refresh']) await self.zypper(['refresh'])
@override
async def _dup(self, download_only: bool) -> None: async def _dup(self, download_only: bool) -> None:
args = ['dup'] args = ['dup']
if download_only: if download_only:
@ -53,6 +61,7 @@ class Distro(Base):
args += self.__update_extra_opts args += self.__update_extra_opts
await self.zypper(args) await self.zypper(args)
@override
async def _reboot_required(self, verbose: bool) -> bool: async def _reboot_required(self, verbose: bool) -> bool:
opts = [] opts = []
if not verbose: if not verbose:
@ -60,26 +69,31 @@ class Distro(Base):
# opts.append('--quiet') # opts.append('--quiet')
opts.append('needs-rebooting') opts.append('needs-rebooting')
ret = await self.zypper(opts, sudo = False, verbose = verbose) ret = await self.zypper(opts, sudo = False, verbose = verbose)
if ret != 0: if ret.status != 0:
return True return True
return False return False
async def _select_by_name(self, names: Iterable[str]) -> Iterable[Package]: @override
async def _select_by_name(self, names: Collection[str]) -> Iterable[Package]:
return await query_packages(names, ec = self.ctx) return await query_packages(names, ec = self.ctx)
@override
async def _install(self, names: Iterable[str], only_update: bool) -> None: async def _install(self, names: Iterable[str], only_update: bool) -> None:
cmd = ['update' if only_update else 'install'] cmd = ['update' if only_update else 'install']
cmd += self.__update_extra_opts cmd += self.__update_extra_opts
cmd += names cmd += names
await self.zypper(cmd) await self.zypper(cmd)
@override
async def _install_local_files( async def _install_local_files(
self, paths: Iterable[str], only_update: bool self, paths: Iterable[str], only_update: bool
) -> None: ) -> None:
await self.rpm(['-U', '--reinstall', *paths]) await self.rpm(['-U', '--reinstall', *paths])
@override
async def _delete(self, names: Iterable[str]) -> None: async def _delete(self, names: Iterable[str]) -> None:
await self.rpm(['-e', *names], sudo = True) await self.rpm(['-e', *names], sudo = True)
@override
async def _pkg_files(self, name: str) -> Iterable[str]: async def _pkg_files(self, name: str) -> Iterable[str]:
return await list_files(name, ec = self.ctx) return await list_files(name, ec = self.ctx)

View file

@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import Any, override, TYPE_CHECKING
from ..FileContext import FileContext as Base from ..FileContext import FileContext as Base
@ -13,7 +13,11 @@ if TYPE_CHECKING:
class Curl(Base): class Curl(Base):
def __init__( def __init__(
self, uri: str | Uri, *args, ec: ExecContext | None = None, **kwargs self,
uri: str | Uri,
*args: Any,
ec: ExecContext | None = None,
**kwargs: Any
) -> None: ) -> None:
def __local() -> Local: def __local() -> Local:
@ -28,6 +32,7 @@ class Curl(Base):
self.__ec = ec if ec else __local() self.__ec = ec if ec else __local()
@override
async def _get( async def _get(
self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str self, path: str, wd: str | None, throw: bool, verbose: bool | None, title: str
) -> Result: ) -> Result:

View file

@ -7,24 +7,27 @@ import pwd
import sys import sys
from functools import cache from functools import cache
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, override
from ..base import Result, StatResult from ..base import Result, StatResult
from ..ExecContext import ExecContext as Base from ..ExecContext import ExecContext as Base
from ..log import ERR, NOTICE, log from ..log import ERR, NOTICE, log
if TYPE_CHECKING: if TYPE_CHECKING:
from typing import Any, Callable
from ..Uri import Uri from ..Uri import Uri
class Local(Base): class Local(Base):
def __init__(self, uri: str | Uri = 'local', *args, **kwargs) -> None: def __init__(self, uri: str | Uri = 'local', *args: Any, **kwargs: Any) -> None:
super().__init__(uri, *args, **kwargs) super().__init__(uri, *args, **kwargs)
@cache @cache
def _username(self) -> str: def _username(self) -> str:
return pwd.getpwuid(os.getuid()).pw_name return pwd.getpwuid(os.getuid()).pw_name
@override
async def _run( async def _run(
self, self,
cmd: list[str], cmd: list[str],
@ -36,13 +39,16 @@ class Local(Base):
log_prefix: str, log_prefix: str,
) -> Result: ) -> Result:
def __log(prio, *args, verbose = verbose): def __log(prio: int, *args: Any, verbose: bool = verbose) -> None:
if verbose: if verbose:
log(prio, log_prefix, *args) log(prio, log_prefix, *args)
def __make_pty_reader(collector: list[bytes], enc_for_verbose: str): def __make_pty_reader(
collector: list[bytes],
enc_for_verbose: str,
) -> Callable[[int], bytes]:
def _read(fd): def _read(fd: int) -> bytes:
ret = os.read(fd, 1024) ret = os.read(fd, 1024)
if not ret: if not ret:
return ret return ret
@ -62,7 +68,7 @@ class Local(Base):
if interactive: if interactive:
import pty import pty
def _spawn(): def _spawn() -> int:
# Apply env in PTY mode by temporarily updating os.environ # Apply env in PTY mode by temporarily updating os.environ
# around spawn. # around spawn.
if mod_env: if mod_env:
@ -112,7 +118,9 @@ class Local(Base):
stdout_log_enc = sys.stdout.encoding or 'utf-8' stdout_log_enc = sys.stdout.encoding or 'utf-8'
stderr_log_enc = sys.stderr.encoding or 'utf-8' stderr_log_enc = sys.stderr.encoding or 'utf-8'
async def read_stream(stream, prio, collector: list[bytes], log_enc: str): async def read_stream(
stream: Any, prio: int, collector: list[bytes], log_enc: str
) -> None:
buf = b'' buf = b''
while True: while True:
chunk = await stream.read(4096) chunk = await stream.read(4096)
@ -155,9 +163,11 @@ class Local(Base):
if cwd is not None: if cwd is not None:
os.chdir(cwd) os.chdir(cwd)
@override
async def _unlink(self, path: str) -> None: async def _unlink(self, path: str) -> None:
os.unlink(path) os.unlink(path)
@override
async def _erase(self, path: str) -> None: async def _erase(self, path: str) -> None:
if os.path.isdir(path): if os.path.isdir(path):
import shutil import shutil
@ -165,26 +175,33 @@ class Local(Base):
return return
os.unlink(path) os.unlink(path)
@override
async def _rename(self, src: str, dst: str) -> None: async def _rename(self, src: str, dst: str) -> None:
os.rename(src, dst) os.rename(src, dst)
@override
async def _mkdir(self, path: str, mode: int) -> None: async def _mkdir(self, path: str, mode: int) -> None:
os.mkdir(path, mode) os.mkdir(path, mode)
@override
async def _stat(self, path: str, follow_symlinks: bool) -> StatResult: async def _stat(self, path: str, follow_symlinks: bool) -> StatResult:
return StatResult.from_os(os.stat(path, follow_symlinks = follow_symlinks)) return StatResult.from_os(os.stat(path, follow_symlinks = follow_symlinks))
@override
async def _file_exists(self, path: str) -> bool: async def _file_exists(self, path: str) -> bool:
return os.path.exists(path) return os.path.exists(path)
@override
async def _chown(self, path: str, owner: str | None, group: str | None) -> None: async def _chown(self, path: str, owner: str | None, group: str | None) -> None:
uid = pwd.getpwnam(owner).pw_uid if owner else -1 uid = pwd.getpwnam(owner).pw_uid if owner else -1
gid = grp.getgrnam(group).gr_gid if group else -1 gid = grp.getgrnam(group).gr_gid if group else -1
os.chown(path, uid, gid) os.chown(path, uid, gid)
@override
async def _chmod(self, path: str, mode: int) -> None: async def _chmod(self, path: str, mode: int) -> None:
os.chmod(path, mode) os.chmod(path, mode)
@override
async def _is_dir(self, path: str, follow_symlinks: bool) -> bool: async def _is_dir(self, path: str, follow_symlinks: bool) -> bool:
if (not follow_symlinks) and os.path.islink(path): if (not follow_symlinks) and os.path.islink(path):
return False return False

View file

@ -5,7 +5,7 @@ import os
import pwd import pwd
from enum import Flag, auto from enum import Flag, auto
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, Any, cast, override
from ..ExecContext import ExecContext from ..ExecContext import ExecContext
from ..log import DEBUG, ERR, INFO, NOTICE, get_caller_pos, log, log_m from ..log import DEBUG, ERR, INFO, NOTICE, get_caller_pos, log, log_m
@ -22,7 +22,9 @@ class SSHClient(ExecContext):
ModEnv = auto() ModEnv = auto()
Wd = auto() Wd = auto()
def __init__(self, uri: Uri | str, caps: Caps = Caps(0), *args, **kwargs) -> None: def __init__(
self, uri: Uri | str, caps: Caps = Caps(0), *args: Any, **kwargs: Any
) -> None:
uri = Uri.pimp(uri) uri = Uri.pimp(uri)
if uri.username is None: if uri.username is None:
uri.set_username(pwd.getpwuid(os.getuid()).pw_name) uri.set_username(pwd.getpwuid(os.getuid()).pw_name)
@ -42,6 +44,7 @@ class SSHClient(ExecContext):
) -> Result: ) -> Result:
pass pass
@override
async def _run( async def _run(
self, self,
cmd: list[str], cmd: list[str],
@ -53,13 +56,13 @@ class SSHClient(ExecContext):
log_prefix: str, log_prefix: str,
) -> Result: ) -> Result:
def __log(prio: int, *args, **kwargs): def __log(prio: int, *args: Any, **kwargs: Any) -> None:
caller = kwargs.get('caller') caller = kwargs.get('caller')
if caller is None: if caller is None:
kwargs['caller'] = get_caller_pos(1) kwargs['caller'] = get_caller_pos(1)
log(prio, log_prefix, *args, **kwargs) log(prio, log_prefix, *args, **kwargs)
def __log_block(prio: int, title: str, block: str | None): def __log_block(prio: int, title: str, block: str | None) -> None:
if self.__caps & self.Caps.LogOutput: if self.__caps & self.Caps.LogOutput:
return return
if block is None: if block is None:
@ -102,6 +105,7 @@ class SSHClient(ExecContext):
return self.uri.port return self.uri.port
@property @property
@override
def username(self) -> str | None: def username(self) -> str | None:
return self.uri.username return self.uri.username
@ -110,8 +114,10 @@ class SSHClient(ExecContext):
return self.uri.password return self.uri.password
def ssh_client( # export def ssh_client( # export
*args, type: str | list[str] | None = None, **kwargs *args: Any,
) -> SSHClient: type: str | list[str] | None = None,
**kwargs: Any
) -> 'SSHClient':
from importlib import import_module from importlib import import_module
errors: list[str] = [] errors: list[str] = []
@ -128,7 +134,7 @@ def ssh_client( # export
ret = getattr(import_module(f'jw.pkg.lib.ec.ssh.{name}'), ret = getattr(import_module(f'jw.pkg.lib.ec.ssh.{name}'),
name)(*args, **kwargs) name)(*args, **kwargs)
log(INFO, f'Using SSH-client "{name}"') log(INFO, f'Using SSH-client "{name}"')
return ret return cast('SSHClient', ret)
except Exception as e: except Exception as e:
msg = f"Can't instantiate SSH client class {name} ({str(e)})" msg = f"Can't instantiate SSH client class {name} ({str(e)})"
errors.append(msg) errors.append(msg)

View file

@ -5,7 +5,11 @@ import shutil
import signal import signal
import sys import sys
import asyncssh # type: ignore[import-not-found] from typing import Any, override
import asyncssh # type: ignore[import-not-found, unused-ignore]
from asyncssh import SSHReader # type: ignore[import-not-found, unused-ignore]
from ...base import Result from ...base import Result
from ...log import DEBUG, ERR, NOTICE, log from ...log import DEBUG, ERR, NOTICE, log
@ -21,10 +25,10 @@ class AsyncSSH(Base):
uri: str, uri: str,
*, *,
client_keys: list[str] | None = None, client_keys: list[str] | None = None,
known_hosts = _USE_DEFAULT_KNOWN_HOSTS, known_hosts: Any = _USE_DEFAULT_KNOWN_HOSTS,
term_type: str | None = None, term_type: str | None = None,
connect_timeout: float | None = 30.0, connect_timeout: float | None = 30.0,
**kwargs, **kwargs: Any,
) -> None: ) -> None:
super().__init__( super().__init__(
@ -42,10 +46,12 @@ class AsyncSSH(Base):
self.__connect_timeout = connect_timeout self.__connect_timeout = connect_timeout
self.__conn: asyncssh.SSHClientConnection | None = None self.__conn: asyncssh.SSHClientConnection | None = None
@override
async def _open(self) -> None: async def _open(self) -> None:
await super()._open() await super()._open()
await self._conn await self._conn
@override
async def _close(self) -> None: async def _close(self) -> None:
if self.__conn is not None: if self.__conn is not None:
try: try:
@ -55,8 +61,8 @@ class AsyncSSH(Base):
log(DEBUG, f'Failed to close connection ({str(e)}, ignored)') log(DEBUG, f'Failed to close connection ({str(e)}, ignored)')
self.__conn = None self.__conn = None
def _connect_kwargs(self, hide_secrets: bool = False) -> dict: def _connect_kwargs(self, hide_secrets: bool = False) -> dict[str, Any]:
kwargs: dict = { kwargs: dict[str, Any] = {
'host': self.hostname, 'host': self.hostname,
'port': self.port, 'port': self.port,
'username': self.username, 'username': self.username,
@ -118,8 +124,8 @@ class AsyncSSH(Base):
async def _read_stream( async def _read_stream(
self, self,
stream, stream: SSHReader[bytes],
prio, prio: int,
collector: list[bytes], collector: list[bytes],
*, *,
verbose: bool, verbose: bool,
@ -207,9 +213,8 @@ class AsyncSSH(Base):
except (BrokenPipeError, OSError): except (BrokenPipeError, OSError):
pass pass
return return
if proc.stdin is None: if proc.stdin is not None:
return proc.stdin.write(data)
proc.stdin.write(data)
await proc.stdin.drain() await proc.stdin.drain()
async def _pump_stdout() -> None: async def _pump_stdout() -> None:
@ -220,7 +225,7 @@ class AsyncSSH(Base):
stdout_parts.append(chunk) stdout_parts.append(chunk)
_write_local(chunk) _write_local(chunk)
def _on_winch(*_args) -> None: def _on_winch(*_args: Any) -> None:
try: try:
proc.change_terminal_size(*self._get_local_term_size()) proc.change_terminal_size(*self._get_local_term_size())
@ -355,6 +360,7 @@ class AsyncSSH(Base):
stdout = b''.join(stdout_parts) if stdout_parts else None stdout = b''.join(stdout_parts) if stdout_parts else None
return Result(stdout, None, exit_code, cmd = cmd) return Result(stdout, None, exit_code, cmd = cmd)
@override
async def _run_ssh( async def _run_ssh(
self, self,
cmd: list[str], cmd: list[str],

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import os import os
from typing import TYPE_CHECKING from typing import Any, override, TYPE_CHECKING
from ...base import InputMode from ...base import InputMode
from ...util import run_cmd from ...util import run_cmd
@ -14,12 +14,12 @@ if TYPE_CHECKING:
class Exec(Base): class Exec(Base):
def __init__(self, uri, *args, **kwargs) -> None: def __init__(self, uri: Any, *args: Any, **kwargs: Any) -> None:
self.__askpass: str | None = None self.__askpass: str | None = None
self.__askpass_orig: dict[str, str | None] = dict() self.__askpass_orig: dict[str, str | None] = dict()
super().__init__(uri = uri, caps = self.Caps.ModEnv, **kwargs) super().__init__(uri = uri, caps = self.Caps.ModEnv, **kwargs)
def __del__(self): def __del__(self) -> None:
for key, val in self.__askpass_orig.items(): for key, val in self.__askpass_orig.items():
if val is None: if val is None:
del os.environ[key] del os.environ[key]
@ -28,7 +28,7 @@ class Exec(Base):
if self.__askpass is not None: if self.__askpass is not None:
os.remove(self.__askpass) os.remove(self.__askpass)
def __init_askpass(self): def __init_askpass(self) -> None:
if self.__askpass is None and self.password is not None: if self.__askpass is None and self.password is not None:
import sys import sys
import tempfile import tempfile
@ -48,6 +48,7 @@ class Exec(Base):
self.__askpass_orig[key] = os.getenv(key) self.__askpass_orig[key] = os.getenv(key)
os.environ[key] = val os.environ[key] = val
@override
async def _run_ssh( async def _run_ssh(
self, self,
cmd: list[str], cmd: list[str],

View file

@ -1,12 +1,12 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING from typing import Any, override, TYPE_CHECKING
# Tolerate missing paramiko imports. jw-pkg is designed to work with what it # Tolerate missing paramiko imports. jw-pkg is designed to work with what it
# finds. # finds.
import paramiko # type: ignore[import-untyped,import-not-found] # error: Library stubs not installed for "paramiko" import paramiko # type: ignore[import-untyped,import-not-found, unused-ignore] # error: Library stubs not installed for "paramiko"
import paramiko.agent # type: ignore[import-untyped,import-not-found] import paramiko.agent # type: ignore[import-untyped,import-not-found, unused-ignore]
import paramiko.SCPClient # type: ignore[import-untyped,import-not-found] import paramiko.SCPClient # type: ignore[import-untyped,import-not-found, unused-ignore]
from ...base import Result from ...base import Result
from ...log import ERR, log from ...log import ERR, log
@ -18,7 +18,7 @@ if TYPE_CHECKING:
class Paramiko(Base): class Paramiko(Base):
def __init__(self, uri, *args, **kwargs) -> None: def __init__(self, uri: Any, *args: Any, **kwargs: Any) -> None:
kwargs['caps'] = (self.Caps.ModEnv, ) kwargs['caps'] = (self.Caps.ModEnv, )
super().__init__(uri, *args, **kwargs) super().__init__(uri, *args, **kwargs)
self.__timeout: float | None = None # Untested self.__timeout: float | None = None # Untested
@ -52,15 +52,18 @@ class Paramiko(Base):
def __scp(self) -> Any: def __scp(self) -> Any:
return paramiko.SCPClient(self.__client.get_transport()) return paramiko.SCPClient(self.__client.get_transport())
@override
async def _open(self) -> None: async def _open(self) -> None:
await super()._open() await super()._open()
self.__client self.__client
@override
async def _close(self) -> None: async def _close(self) -> None:
if self.___client is not None: if self.___client is not None:
self.___client.close() self.___client.close()
self.___client = None self.___client = None
@override
async def _run_ssh( async def _run_ssh(
self, self,
cmd: list[str], cmd: list[str],

View file

@ -42,8 +42,6 @@ def join_cmd(
""" """
ret: list[str] = [] ret: list[str] = []
for token in cmd: for token in cmd:
if not isinstance(token, str):
token = str(token)
if token in operators: if token in operators:
ret.append(token) ret.append(token)
else: else:

View file

@ -7,7 +7,7 @@ import syslog
from datetime import datetime from datetime import datetime
from os.path import basename from os.path import basename
from typing import TYPE_CHECKING from typing import TYPE_CHECKING, cast
if TYPE_CHECKING: if TYPE_CHECKING:
import io import io
@ -126,7 +126,7 @@ _prio_colors = {
class Stream: class Stream:
def __init__(self, stream, flags): def __init__(self, stream: Any, flags: int):
self.stream = stream self.stream = stream
self.flags = flags self.flags = flags
@ -142,12 +142,12 @@ def pad(token: str, total_size: int, right_align: bool = False) -> str:
return space + token return space + token
return token + space return token + space
def add_capture_stream(stream, flags = 0x0): def add_capture_stream(stream: Any, flags: int = 0x0) -> int:
ret = _stream_descriptors.pop() ret = _stream_descriptors.pop()
_streams[ret] = Stream(stream = stream, flags = flags) _streams[ret] = Stream(stream = stream, flags = flags)
return ret return ret
def rm_capture_stream(sd): def rm_capture_stream(sd: int) -> None:
del _streams[sd] del _streams[sd]
_stream_descriptors.append(sd) _stream_descriptors.append(sd)
@ -166,17 +166,17 @@ def get_caller_pos(up: int = 1,
if kwargs and 'caller' in kwargs: if kwargs and 'caller' in kwargs:
r = kwargs['caller'] r = kwargs['caller']
del kwargs['caller'] del kwargs['caller']
return r return cast('Tuple[str, str, int]', r)
caller = inspect.stack()[up + 1] caller = inspect.stack()[up + 1]
mod = inspect.getmodule(caller[0]) mod = inspect.getmodule(caller[0])
mod_name = '' if mod is None else mod.__name__ mod_name = '' if mod is None else mod.__name__
return (mod_name, basename(caller.filename), caller.lineno) return (mod_name, basename(caller.filename), caller.lineno)
def log_m(prio: int, *args, **kwargs) -> None: # export def log_m(prio: int, *args: Any, **kwargs: Any) -> None: # export
if prio > _level: if prio > _level:
return return
margs = ''
if len(args): if len(args):
margs = ''
for a in args: for a in args:
if isinstance(a, list): if isinstance(a, list):
margs += '\n'.join([str(elem) for elem in a]) margs += '\n'.join([str(elem) for elem in a])
@ -190,7 +190,12 @@ def log_m(prio: int, *args, **kwargs) -> None: # export
for line in margs[1:].split('\n'): for line in margs[1:].split('\n'):
log(prio, line, **kwargs, caller = caller) log(prio, line, **kwargs, caller = caller)
def log(prio: int, *args, only_printable: bool = False, **kwargs) -> None: # export def log( # export
prio: int,
*args: Any,
only_printable: bool = False,
**kwargs: Any
) -> None:
if prio > _level: if prio > _level:
return return
@ -259,7 +264,12 @@ def log(prio: int, *args, only_printable: bool = False, **kwargs) -> None: # ex
for file in files: for file in files:
print(msg, file = file) print(msg, file = file)
def throw(*args, prio = ERR, caller = None, **kwargs) -> None: def throw(
*args: Any,
prio: int = ERR,
caller: Tuple[str, str, int] | None = None,
**kwargs: Any
) -> None:
if caller is None: if caller is None:
caller = get_caller_pos(1) caller = get_caller_pos(1)
msg = ' '.join([str(arg) for arg in args]) msg = ' '.join([str(arg) for arg in args])
@ -331,7 +341,7 @@ def append_to_prefix(prefix: str) -> str: # export
_clean_log_prefix = _clean_str_regex.sub('', _log_prefix) _clean_log_prefix = _clean_str_regex.sub('', _log_prefix)
return r return r
def remove_from_prefix(count) -> str: # export def remove_from_prefix(count: int | str) -> str: # export
if isinstance(count, str): if isinstance(count, str):
count = len(count) count = len(count)
global _log_prefix global _log_prefix

View file

@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING, Iterable from typing import TYPE_CHECKING, Iterable, cast
if TYPE_CHECKING: if TYPE_CHECKING:
from ..ExecContext import ExecContext from ..ExecContext import ExecContext
@ -11,7 +11,7 @@ from ..util import run_cmd, run_sudo
_meta_map: dict[str, str] | None = None _meta_map: dict[str, str] | None = None
def meta_map(): def meta_map() -> dict[str, str]:
global _meta_map global _meta_map
if _meta_map is None: if _meta_map is None:
_meta_map = Package.order_tags( _meta_map = Package.order_tags(
@ -59,7 +59,7 @@ async def query_packages(names: Iterable[str] = [],
) )
# dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n' # dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n'
specs = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo = False, ec = ec) specs = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo = False, ec = ec)
return Package.parse_specs_str(specs) return cast('Iterable[Package]', Package.parse_specs_str(specs))
async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]: async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]:
file_list_str = await run_dpkg(['-L', pkg], sudo = False, ec = ec) file_list_str = await run_dpkg(['-L', pkg], sudo = False, ec = ec)

View file

@ -1,6 +1,6 @@
from __future__ import annotations from __future__ import annotations
from typing import TYPE_CHECKING, Iterable from typing import Any, TYPE_CHECKING, Collection, Iterable, cast
from ..base import InputMode from ..base import InputMode
from ..Package import Package from ..Package import Package
@ -11,7 +11,7 @@ if TYPE_CHECKING:
_meta_map: dict[str, str] | None = None _meta_map: dict[str, str] | None = None
def meta_map(): def meta_map() -> dict[str, str]:
global _meta_map global _meta_map
if _meta_map is None: if _meta_map is None:
_meta_map = Package.order_tags( _meta_map = Package.order_tags(
@ -30,7 +30,7 @@ async def run_rpm( # export
sudo: bool = False, sudo: bool = False,
ec: ExecContext | None = None, ec: ExecContext | None = None,
mode: InputMode = InputMode.OptInteractive, mode: InputMode = InputMode.OptInteractive,
**kwargs, **kwargs: Any,
) -> str: ) -> str:
cmd = ['/usr/bin/rpm'] cmd = ['/usr/bin/rpm']
cmd.extend(args) cmd.extend(args)
@ -41,7 +41,7 @@ async def run_rpm( # export
return result.stdout_str return result.stdout_str
async def query_packages( # export async def query_packages( # export
names: Iterable[str] = [], names: Collection[str] = [],
ec: ExecContext | None = None, ec: ExecContext | None = None,
) -> Iterable[Package]: ) -> Iterable[Package]:
fmt_str = ( fmt_str = (
@ -58,7 +58,7 @@ async def query_packages( # export
mode = InputMode.NonInteractive, mode = InputMode.NonInteractive,
ec = ec ec = ec
) )
return Package.parse_specs_str(specs) return cast('Iterable[Package]', Package.parse_specs_str(specs))
async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]: async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]:
stdout = await run_rpm( stdout = await run_rpm(

View file

@ -5,7 +5,7 @@ import os
import sys import sys
from enum import Enum, auto from enum import Enum, auto
from typing import TYPE_CHECKING, Iterable, TypeVar, cast from typing import TYPE_CHECKING, Any, Iterable, TypeVar, cast
from .base import Input, InputMode, Result from .base import Input, InputMode, Result
from .log import DEBUG, ERR, log from .log import DEBUG, ERR, log
@ -23,7 +23,7 @@ class AskpassKey(Enum):
Username = auto() Username = auto()
Password = auto() Password = auto()
def pretty_cmd(cmd: list[str] | None = None, wd = None): def pretty_cmd(cmd: list[str] | None = None, wd: str | None = None) -> str:
if cmd is None: if cmd is None:
cmd = sys.argv cmd = sys.argv
tokens = [cmd[0]] tokens = [cmd[0]]
@ -38,11 +38,11 @@ def pretty_cmd(cmd: list[str] | None = None, wd = None):
# See ExecContext.run() for what this function does # See ExecContext.run() for what this function does
async def run_cmd( async def run_cmd(
*args, *args: Any,
ec: ExecContext | None = None, ec: ExecContext | None = None,
verbose: bool | None = None, verbose: bool | None = None,
cmd_input: Input = InputMode.NonInteractive, cmd_input: Input = InputMode.NonInteractive,
**kwargs, **kwargs: Any,
) -> Result: ) -> Result:
if verbose is None: if verbose is None:
verbose = False if ec is None else ec.verbose_default verbose = False if ec is None else ec.verbose_default
@ -56,12 +56,12 @@ async def run_cmd(
async def run_curl( async def run_curl(
args: list[str], args: list[str],
wd = None, wd: str | None = None,
throw = None, throw: bool | None = None,
verbose = None, verbose: bool | None = None,
cmd_input = InputMode.NonInteractive, cmd_input: Input = InputMode.NonInteractive,
ec: ExecContext | None = None, ec: ExecContext | None = None,
decode = False, decode: bool = False,
) -> Result: ) -> Result:
if verbose is None: if verbose is None:
verbose = False if ec is None else ec.verbose_default verbose = False if ec is None else ec.verbose_default
@ -76,7 +76,7 @@ async def run_curl(
async def run_curl_into( async def run_curl_into(
expected_type: type[T], expected_type: type[T],
args: list[str], args: list[str],
**kwargs, **kwargs: Any,
) -> T: ) -> T:
result = await run_curl(args, **kwargs) result = await run_curl(args, **kwargs)
stdout = result.stdout_str stdout = result.stdout_str
@ -93,7 +93,7 @@ async def run_curl_into(
raise TypeError( raise TypeError(
f'Expected {expected_type.__name__}, got {type(ret).__name__} from Curl' f'Expected {expected_type.__name__}, got {type(ret).__name__} from Curl'
) )
return cast('T', ret) return ret
async def run_askpass( async def run_askpass(
askpass_env: list[str], askpass_env: list[str],
@ -122,6 +122,8 @@ async def run_askpass(
continue # Can't get user name from SSH_ASKPASS continue # Can't get user name from SSH_ASKPASS
case AskpassKey.Password: case AskpassKey.Password:
exe_arg += 'Password' exe_arg += 'Password'
case _:
pass
result = await run_cmd([exe, exe_arg], throw = throw, ec = ec) result = await run_cmd([exe, exe_arg], throw = throw, ec = ec)
if result.status == 0 and result.stdout_or_none is not None: if result.status == 0 and result.stdout_or_none is not None:
ret = result.stdout_str_or_none ret = result.stdout_str_or_none
@ -137,11 +139,11 @@ async def run_askpass(
async def run_sudo( async def run_sudo(
cmd: list[str], cmd: list[str],
*args, *args: Any,
interactive: bool = True, interactive: bool = True,
ec: ExecContext | None = None, ec: ExecContext | None = None,
**kwargs, **kwargs: Any,
): ) -> Result:
if ec is None: if ec is None:
from .ec.Local import Local from .ec.Local import Local
@ -150,10 +152,10 @@ async def run_sudo(
async def get( async def get(
uri: str | Uri, uri: str | Uri,
*args, *args: Any,
ctx: FileContext | None = None, ctx: FileContext | None = None,
content_filter: ProcFilter | list[ProcFilter] | ProcPipeline | None = None, content_filter: ProcFilter | list[ProcFilter] | ProcPipeline | None = None,
**kwargs, **kwargs: Any,
) -> Result: ) -> Result:
uri = Uri.pimp(uri) uri = Uri.pimp(uri)
if ctx is None or uri.id != ctx.uri.id: if ctx is None or uri.id != ctx.uri.id:
@ -170,7 +172,7 @@ async def copy(
owner: str | None = None, owner: str | None = None,
group: str | None = None, group: str | None = None,
mode: int | None = None, mode: int | None = None,
throw = True, throw: bool = True,
) -> Exception | str | list[str]: ) -> Exception | str | list[str]:
if not isinstance(src_uri, str): if not isinstance(src_uri, str):
ret: list[str] = [] ret: list[str] = []
@ -223,7 +225,7 @@ async def get_username( # export
f'Username mismatch: called with --username="{args.username}", ' f'Username mismatch: called with --username="{args.username}", '
f'URL has user name "{url_user}"' f'URL has user name "{url_user}"'
) )
return args.username return str(args.username)
if url_user is not None: if url_user is not None:
return url_user return url_user
return await run_askpass(askpass_env, AskpassKey.Username, ec = ec) return await run_askpass(askpass_env, AskpassKey.Username, ec = ec)
@ -242,7 +244,7 @@ async def get_password( # export
if args is not None and hasattr(args, 'password'): if args is not None and hasattr(args, 'password'):
# use getattr(), because we don't necessarily want to have insecure # use getattr(), because we don't necessarily want to have insecure
# --password among options # --password among options
ret = getattr(args, 'password') ret = cast('str | None', getattr(args, 'password'))
if ret is not None: if ret is not None:
return ret return ret
if url is not None: if url is not None: