janware/jw-pkg
Watch
1
Code Pull requests Activity Actions

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
Conversation 0 Commits 31 Files changed 92 +588 -324
92 changed files with 588 additions and 324 deletions
Download patch file Download diff file Expand all files Collapse all files

12
conf/project/pyrightconfig-base.json
View file

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

13
conf/templates/pyproject.toml
View file

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

62
src/python/jw/pkg/App.py
View file

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

2
src/python/jw/pkg/CmdBase.py
View file

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

6
src/python/jw/pkg/cmds/Cmd.py
View file

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

8
src/python/jw/pkg/cmds/CmdPkg.py
View file

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

8
src/python/jw/pkg/cmds/CmdPlatform.py
View file

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

8
src/python/jw/pkg/cmds/CmdPosix.py
View file

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

8
src/python/jw/pkg/cmds/CmdProjects.py
View file

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

8
src/python/jw/pkg/cmds/CmdSecrets.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdDelete.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdDup.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdInstall.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdLs.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdMeta.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdRebootRequired.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdRefresh.py
View file

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

4
src/python/jw/pkg/cmds/pkg/CmdSelect.py
View file

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

3
src/python/jw/pkg/cmds/pkg/NamedPkgsCmd.py
View file

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

4
src/python/jw/pkg/cmds/platform/CmdInfo.py
View file

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

3
src/python/jw/pkg/cmds/posix/Cmd.py
View file

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

4
src/python/jw/pkg/cmds/posix/CmdCopy.py
View file

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

8
src/python/jw/pkg/cmds/posix/CmdTar.py
View file

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

5
src/python/jw/pkg/cmds/posix/tar/Cmd.py
View file

@ -1,7 +1,7 @@
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING
from typing import Any, override, TYPE_CHECKING
from ....CmdBase import CmdBase as Base
from ....lib.FileContext import FileContext
@ -20,13 +20,14 @@ class Cmd(Base): # export
self.__tar_io: None = None
@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:
ret.src.add_proc_filter(
FileContext.Direction.In, ProcFilterGpg(ec = self.app.exec_context)
)
yield ret
@override
def add_arguments(self, parser: ArgumentParser) -> None:
super().add_arguments(parser)
parser.add_argument(

4
src/python/jw/pkg/cmds/posix/tar/CmdExtract.py
View file

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

6
src/python/jw/pkg/cmds/projects/BaseCmdPkgRelations.py
View file

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

3
src/python/jw/pkg/cmds/projects/Cmd.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdBuild.py
View file

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

6
src/python/jw/pkg/cmds/projects/CmdCanonicalizeRemotes.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdCflags.py
View file

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

8
src/python/jw/pkg/cmds/projects/CmdCheck.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdCommands.py
View file

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

9
src/python/jw/pkg/cmds/projects/CmdCreateFile.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdCreatePkgConfig.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdExepath.py
View file

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

6
src/python/jw/pkg/cmds/projects/CmdGetAuthInfo.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdGetval.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdHtdocsDir.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdLdflags.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdLdlibpath.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdLibname.py
View file

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

6
src/python/jw/pkg/cmds/projects/CmdListRepos.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdModules.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdPath.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdProjDir.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdPythonpath.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdRequiredOsPkg.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdSummary.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdTest.py
View file

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

4
src/python/jw/pkg/cmds/projects/CmdTmplDir.py
View file

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

3
src/python/jw/pkg/cmds/projects/check/Cmd.py
View file

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

4
src/python/jw/pkg/cmds/projects/check/CmdDep.py
View file

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

13
src/python/jw/pkg/cmds/projects/lib/pkg_relations.py
View file

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

17
src/python/jw/pkg/cmds/projects/lib/templates.py
View file

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

17
src/python/jw/pkg/cmds/secrets/Cmd.py
View file

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

4
src/python/jw/pkg/cmds/secrets/CmdCompileTemplates.py
View file

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

4
src/python/jw/pkg/cmds/secrets/CmdInstall.py
View file

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

4
src/python/jw/pkg/cmds/secrets/CmdListCompilationOutput.py
View file

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

4
src/python/jw/pkg/cmds/secrets/CmdListSecrets.py
View file

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

3
src/python/jw/pkg/cmds/secrets/CmdListTemplates.py
View file

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

3
src/python/jw/pkg/cmds/secrets/CmdRmCompilationOutput.py
View file

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

16
src/python/jw/pkg/cmds/secrets/lib/DistroContext.py
View file

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

6
src/python/jw/pkg/cmds/secrets/lib/FilesContext.py
View file

@ -23,7 +23,11 @@ class FilesContext:
def ctx(self) -> FileContext:
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] = {}
try:
result = await self.ctx.get(path)

59
src/python/jw/pkg/lib/App.py
View file

@ -6,7 +6,7 @@ import os
import sys
from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser, Namespace
from typing import TYPE_CHECKING, Any
from typing import Any, cast, override, TYPE_CHECKING
from .AsyncRunner import AsyncRunner
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
if TYPE_CHECKING:
from collections.abc import Awaitable
from collections.abc import Awaitable, Collection
from typing import TypeVar
T = TypeVar('T')
class App: # export
def _add_arguments(self, parser):
def _add_arguments(self, parser: ArgumentParser) -> None:
self.__parser.add_argument(
'--log-flags', help = 'Log flags', default = self.__default_log_flags
)
@ -51,13 +51,16 @@ class App: # export
eloop: None = None,
) -> None:
def add_cmd_to_parser(cmd, parsers):
parser = parsers.add_parser(
cmd.name,
help = cmd.help,
description = cmd.description,
aliases = cmd.aliases,
formatter_class = ArgumentDefaultsHelpFormatter,
def add_cmd_to_parser(cmd: AbstractCmd, parsers: Any) -> ArgumentParser:
parser = cast(
'ArgumentParser',
parsers.add_parser(
cmd.name,
help = cmd.help,
description = cmd.description,
aliases = cmd.aliases,
formatter_class = ArgumentDefaultsHelpFormatter,
)
)
parser.set_defaults(func = cmd.run)
cmd.add_arguments(parser)
@ -67,8 +70,8 @@ class App: # export
def add_cmds_to_parser(
parent: AbstractCmd | App,
parser: ArgumentParser,
cmds,
all = False
cmds: Collection[AbstractCmd],
all: bool = False
) -> None:
if not cmds:
return
@ -150,7 +153,7 @@ class App: # export
cmd_classes: LoadTypes[AbstractCmd] = LoadTypes(
modules if modules else ['__main__'],
type_name_filter = name_filter,
type_filter = [AbstractCmd], # type: ignore[type-abstract]
type_filter = [AbstractCmd],
)
add_all_parsers = (
'-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'
)
def __del__(self):
def __del__(self) -> None:
if self.__own_eloop:
if self.__eloop is not None:
self.__eloop.close()
@ -178,24 +181,27 @@ class App: # export
async def __aenter__(self) -> None:
pass
async def __aexit__(self, exc_type, exc, tb) -> None:
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
pass
async def __run(self, argv = None) -> None:
async def __run(self, argv: list[str] | None = None) -> None:
try:
# Import argcomplete only here to not require it to be compatible
# with minimal environments
from argcomplete.completers import ( # type: ignore[import-not-found]
from argcomplete.completers import ( # type: ignore[import-not-found, unused-ignore]
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
import argcomplete # type: ignore[import-not-found]
import argcomplete # type: ignore[import-not-found, unused-ignore]
argcomplete.autocomplete(self.__parser, default_completer = NoopCompleter())
@ -246,7 +252,7 @@ class App: # export
self.__parser.print_help()
return None
# 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:
return self.async_runner.call(awaitable, timeout)
@ -283,7 +289,7 @@ class App: # export
def parser(self) -> ArgumentParser:
return self.__parser
def run(self, argv = None) -> None:
def run(self, argv: list[str] | None = None) -> None:
try:
ret = self.eloop.run_until_complete(self.__run(argv))
finally:
@ -293,7 +299,10 @@ class App: # export
return ret
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)
return app.run(argv = argv)
app.run(argv = argv)

11
src/python/jw/pkg/lib/AsyncRunner.py
View file

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

25
src/python/jw/pkg/lib/Cmd.py
View file

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

10
src/python/jw/pkg/lib/CopyContext.py
View file

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

33
src/python/jw/pkg/lib/Distro.py
View file

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

25
src/python/jw/pkg/lib/ExecContext.py
View file

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

18
src/python/jw/pkg/lib/FileContext.py
View file

@ -4,7 +4,7 @@ import abc
from enum import Enum, auto
from functools import cached_property
from typing import TYPE_CHECKING
from typing import Any, override, TYPE_CHECKING
from .log import DEBUG, ERR, log
from .Uri import Uri
@ -24,7 +24,7 @@ class FileContext(abc.ABC):
self,
uri: str | Uri,
interactive: bool | None = None,
verbose_default = False,
verbose_default: bool = False,
chroot: bool = False,
in_pipe: ProcPipeline | None = None,
out_pipe: ProcPipeline | None = None,
@ -43,17 +43,18 @@ class FileContext(abc.ABC):
f'= "{verbose_default}"'
)
async def __aenter__(self):
async def __aenter__(self) -> 'FileContext':
await self.open()
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()
@override
def __repr__(self) -> str:
return self.__uri.id
def __pipe(self, d: Direction):
def __pipe(self, d: Direction) -> 'ProcPipeline':
match d:
case self.Direction.In:
if not self.__in_pipe:
@ -75,7 +76,7 @@ class FileContext(abc.ABC):
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)
async def _open(self) -> None:
@ -296,13 +297,12 @@ class FileContext(abc.ABC):
except Exception as e:
log(ERR, f'{self.log_name}: Failed to stat({path}) ({str(e)})')
raise
return False
async def is_dir(self, path: str, 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)
@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)
match uri.protocol:
case 'local' | 'file':

11
src/python/jw/pkg/lib/Package.py
View file

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

3
src/python/jw/pkg/lib/PackageFilter.py
View file

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

3
src/python/jw/pkg/lib/ProcFilter.py
View file

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

3
src/python/jw/pkg/lib/ProcFilterGpg.py
View file

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

9
src/python/jw/pkg/lib/Result.py
View file

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

25
src/python/jw/pkg/lib/TarIo.py
View file

@ -9,14 +9,14 @@ from tarfile import TarFile, TarInfo
from .CopyContext import CopyContext
from .ExecContext import ExecContext
from .log import DEBUG, ERR, log
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Any, override
if TYPE_CHECKING:
from .base import StatResult
class TarIo(CopyContext):
def __init__(self, *args, **kwargs) -> None:
def __init__(self, *args: Any, **kwargs: Any) -> None:
kwargs['chroot'] = False
super().__init__(*args, **kwargs)
@ -44,7 +44,7 @@ class TarIo(CopyContext):
async def _read_filtered(
self,
path,
path: str,
path_filter: list[str] | None = None,
matched: list[str] | None = None,
) -> bytes:
@ -81,7 +81,7 @@ class TarIo(CopyContext):
return ret
@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:
raise NotImplementedError
# return TarIoTarFile(*args, **kwargs)
@ -89,6 +89,7 @@ class TarIo(CopyContext):
class TarIoTarFile(TarIo):
@override
async def _extract(self, blob: bytes, root: str | None = None) -> None:
tf = TarFile(fileobj = io.BytesIO(blob))
for info in tf.getmembers():
@ -111,7 +112,16 @@ class TarIoTarFile(TarIo):
class TarIoTarExec(TarIo):
@override
async def _extract(self, blob: bytes, root: str | None = None) -> None:
cmd = ['tar']
if root is not None:
cmd += ['-C', root]
cmd += ['-x', '-f', '-']
await self.dst.run(cmd, cmd_input = blob)
@property
@override
def dst(self) -> ExecContext:
ret = super().dst
if not isinstance(ret, ExecContext):
@ -120,10 +130,3 @@ class TarIoTarExec(TarIo):
'context, which only has a file context'
)
return ret
async def _extract(self, blob: bytes, root: str | None = None) -> None:
cmd = ['tar']
if root is not None:
cmd += ['-C', root]
cmd += ['-x', '-f', '-']
await self.dst.run(cmd, cmd_input = blob)

15
src/python/jw/pkg/lib/Types.py
View file

@ -5,7 +5,7 @@ import os
import re
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
@ -17,6 +17,7 @@ T = TypeVar('T')
class Types(abc.ABC, Iterable[type[T]], Generic[T]): # export
@override
def __iter__(self) -> Iterator[type[T]]:
return iter(self._classes())
@ -32,7 +33,7 @@ class Types(abc.ABC, Iterable[type[T]], Generic[T]): # export
def _stringify(self) -> list[str]:
pass
def dump(self, prio: int, *args, **kwargs) -> None:
def dump(self, prio: int, *args: Any, **kwargs: Any) -> None:
contents = self._stringify()
log(prio, ',--- ', *args, **kwargs)
for line in contents:
@ -46,8 +47,8 @@ class LoadTypes(Types[T]): # export
mod_names: Iterable[str],
type_name_filter: str | None = None,
type_filter: Sequence[type[Any]] | None = None,
debug_level = None,
):
debug_level: int | None = None,
) -> None:
if debug_level is None:
val = os.getenv('JW_LOG_LEVEL_LOAD_TYPES')
if val is not None:
@ -60,11 +61,12 @@ class LoadTypes(Types[T]): # export
self.__mod_names = mod_names
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:
log(self.__debug_level, *args, **kwargs)
def _stringify(self):
@override
def _stringify(self) -> list[str]:
tf = 'None' if self.__type_filter is None else (
', '.join([str(f) for f in self.__type_filter])
)
@ -74,6 +76,7 @@ class LoadTypes(Types[T]): # export
'mod_names: ' + ', '.join(self.__mod_names),
]
@override
def _classes(self) -> Iterable[type[T]]:
if self.__classes is None:

4
src/python/jw/pkg/lib/Uri.py
View file

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

11
src/python/jw/pkg/lib/distros/arch/Distro.py
View file

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

13
src/python/jw/pkg/lib/distros/debian/Distro.py
View file

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

24
src/python/jw/pkg/lib/distros/suse/Distro.py
View file

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

9
src/python/jw/pkg/lib/ec/Curl.py
View file

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

31
src/python/jw/pkg/lib/ec/Local.py
View file

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

20
src/python/jw/pkg/lib/ec/SSHClient.py
View file

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

28
src/python/jw/pkg/lib/ec/ssh/AsyncSSH.py
View file

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

9
src/python/jw/pkg/lib/ec/ssh/Exec.py
View file

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

13
src/python/jw/pkg/lib/ec/ssh/Paramiko.py
View file

@ -1,12 +1,12 @@
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
# finds.
import paramiko # type: ignore[import-untyped,import-not-found] # error: Library stubs not installed for "paramiko"
import paramiko.agent # type: ignore[import-untyped,import-not-found]
import paramiko.SCPClient # type: ignore[import-untyped,import-not-found]
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, unused-ignore]
import paramiko.SCPClient # type: ignore[import-untyped,import-not-found, unused-ignore]
from ...base import Result
from ...log import ERR, log
@ -18,7 +18,7 @@ if TYPE_CHECKING:
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, )
super().__init__(uri, *args, **kwargs)
self.__timeout: float | None = None # Untested
@ -52,15 +52,18 @@ class Paramiko(Base):
def __scp(self) -> Any:
return paramiko.SCPClient(self.__client.get_transport())
@override
async def _open(self) -> None:
await super()._open()
self.__client
@override
async def _close(self) -> None:
if self.___client is not None:
self.___client.close()
self.___client = None
@override
async def _run_ssh(
self,
cmd: list[str],

2
src/python/jw/pkg/lib/ec/ssh/util.py
View file

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

30
src/python/jw/pkg/lib/log.py
View file

@ -7,7 +7,7 @@ import syslog
from datetime import datetime
from os.path import basename
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, cast
if TYPE_CHECKING:
import io
@ -126,7 +126,7 @@ _prio_colors = {
class Stream:
def __init__(self, stream, flags):
def __init__(self, stream: Any, flags: int):
self.stream = stream
self.flags = flags
@ -142,12 +142,12 @@ def pad(token: str, total_size: int, right_align: bool = False) -> str:
return space + token
return token + space
def add_capture_stream(stream, flags = 0x0):
def add_capture_stream(stream: Any, flags: int = 0x0) -> int:
ret = _stream_descriptors.pop()
_streams[ret] = Stream(stream = stream, flags = flags)
return ret
def rm_capture_stream(sd):
def rm_capture_stream(sd: int) -> None:
del _streams[sd]
_stream_descriptors.append(sd)
@ -166,17 +166,17 @@ def get_caller_pos(up: int = 1,
if kwargs and 'caller' in kwargs:
r = kwargs['caller']
del kwargs['caller']
return r
return cast('Tuple[str, str, int]', r)
caller = inspect.stack()[up + 1]
mod = inspect.getmodule(caller[0])
mod_name = '' if mod is None else mod.__name__
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:
return
margs = ''
if len(args):
margs = ''
for a in args:
if isinstance(a, list):
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'):
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:
return
@ -259,7 +264,12 @@ def log(prio: int, *args, only_printable: bool = False, **kwargs) -> None: # ex
for file in files:
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:
caller = get_caller_pos(1)
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)
return r
def remove_from_prefix(count) -> str: # export
def remove_from_prefix(count: int | str) -> str: # export
if isinstance(count, str):
count = len(count)
global _log_prefix

6
src/python/jw/pkg/lib/pm/dpkg.py
View file

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

10
src/python/jw/pkg/lib/pm/rpm.py
View file

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

40
src/python/jw/pkg/lib/util.py
View file

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