ahriman/src/ahriman/application/handlers/status.py

71 lines
2.7 KiB
Python

#
# Copyright (c) 2021-2023 ahriman team.
#
# This file is part of ahriman
# (see https://github.com/arcan1s/ahriman).
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import argparse
from collections.abc import Callable
from ahriman.application.application import Application
from ahriman.application.handlers import Handler
from ahriman.core.configuration import Configuration
from ahriman.core.formatters import PackagePrinter, StatusPrinter
from ahriman.models.build_status import BuildStatus
from ahriman.models.package import Package
from ahriman.models.repository_id import RepositoryId
class Status(Handler):
"""
package status handler
"""
ALLOW_MULTI_ARCHITECTURE_RUN = False # conflicting io
@classmethod
def run(cls, args: argparse.Namespace, repository_id: RepositoryId, configuration: Configuration, *,
report: bool) -> None:
"""
callback for command line
Args:
args(argparse.Namespace): command line args
repository_id(RepositoryId): repository unique identifier
configuration(Configuration): configuration instance
report(bool): force enable or disable reporting
"""
# we are using reporter here
client = Application(repository_id, configuration, report=True).repository.reporter
if args.ahriman:
service_status = client.status_get()
StatusPrinter(service_status.status).print(verbose=args.info)
if args.package:
packages: list[tuple[Package, BuildStatus]] = sum(
(client.package_get(base) for base in args.package),
start=[])
else:
packages = client.package_get(None)
Status.check_if_empty(args.exit_code, not packages)
comparator: Callable[[tuple[Package, BuildStatus]], str] = lambda item: item[0].base
filter_fn: Callable[[tuple[Package, BuildStatus]], bool] =\
lambda item: args.status is None or item[1].status == args.status
for package, package_status in sorted(filter(filter_fn, packages), key=comparator):
PackagePrinter(package, package_status).print(verbose=args.info)