cli: deduplicate typical messages
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Anna “CyberTailor” 2024-01-10 09:51:00 +05:00
parent ffe962d4a5
commit 768d711be0
Signed by: CyberTaIlor
GPG Key ID: E7B76EDC50864BB1
3 changed files with 46 additions and 16 deletions

View File

@ -8,6 +8,7 @@ import threading
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import field
from enum import Enum, auto
from typing import Any
import click
@ -73,6 +74,16 @@ class BugzillaOptions:
chronological_sort: bool = False
class Message(Enum):
""" Typical messages. """
CACHE_READ = auto()
CACHE_LOAD = auto()
CACHE_WRITE = auto()
EMPTY_RESPONSE = auto()
NO_WORK = auto()
@dataclass
class Options:
""" Global options. """
@ -116,3 +127,24 @@ class Options:
"""
kwargs.pop("color", None)
click.secho(*args, color=self.colors, **kwargs) # type: ignore
def say(self, msgid: Message) -> None:
"""
Output one of pre-configured messages to the terminal.
:param msgid: message type
"""
match msgid:
case Message.CACHE_LOAD:
self.vecho("Checking for cached data", nl=False, err=True)
case Message.CACHE_READ:
self.vecho("Reading cached data", nl=False, err=True)
case Message.CACHE_WRITE:
self.vecho("Caching data", nl=False, err=True)
case Message.EMPTY_RESPONSE:
self.secho("Hmmm, no data returned. Try again with different "
"arguments.", fg="yellow")
case Message.NO_WORK:
self.secho("Congrats! You have nothing to do!", fg="green")
case _:
raise TypeError(f"Unknown message identifier: {msgid}")

View File

@ -41,7 +41,7 @@ import click
import gentoopm
from tabulate import tabulate
from find_work.cli import Options, ProgressDots
from find_work.cli import Message, Options, ProgressDots
from find_work.constants import BUGZILLA_URL
from find_work.types import BugView
from find_work.utils import (
@ -145,12 +145,12 @@ def outdated(options: Options) -> None:
options.cache_key.feed("outdated")
dots = ProgressDots(options.verbose)
options.vecho("Checking for cached data", nl=False, err=True)
options.say(Message.CACHE_LOAD)
with dots():
cached_data = read_json_cache(options.cache_key,
object_hook=as_datetime)
if cached_data is not None:
options.vecho("Loading cached data", nl=False, err=True)
options.say(Message.CACHE_READ)
with dots():
data = _bugs_from_json(cached_data)
else:
@ -158,16 +158,15 @@ def outdated(options: Options) -> None:
with dots():
data = _fetch_bump_requests(options)
if len(data) == 0:
options.secho("Hmmm, no data returned. Try again with "
"different arguments.", fg="yellow")
options.say(Message.EMPTY_RESPONSE)
return
options.vecho("Caching data", nl=False, err=True)
options.say(Message.CACHE_WRITE)
with dots():
json_data = _bugs_to_json(data)
write_json_cache(json_data, options.cache_key, cls=BugEncoder)
bumps = _collect_bump_requests(data, options)
if len(bumps) == 0:
options.secho("Congrats! You have nothing to do!", fg="green")
else:
if len(bumps) != 0:
options.echo(tabulate(bumps, tablefmt="plain")) # type: ignore
else:
options.say(Message.NO_WORK)

View File

@ -16,7 +16,7 @@ from pydantic import RootModel
from repology_client.types import Package
from sortedcontainers import SortedSet
from find_work.cli import Options, ProgressDots
from find_work.cli import Message, Options, ProgressDots
from find_work.types import VersionBump
from find_work.utils import (
aiohttp_session,
@ -87,11 +87,11 @@ def _collect_version_bumps(data: Iterable[set[Package]],
async def _outdated(options: Options) -> None:
dots = ProgressDots(options.verbose)
options.vecho("Checking for cached data", nl=False, err=True)
options.say(Message.CACHE_LOAD)
with dots():
cached_data = read_json_cache(options.cache_key)
if cached_data is not None:
options.vecho("Loading cached data", nl=False, err=True)
options.say(Message.CACHE_READ)
with dots():
data = _projects_from_json(cached_data)
else:
@ -100,10 +100,9 @@ async def _outdated(options: Options) -> None:
with dots():
data = await _fetch_outdated(options)
except repology_client.exceptions.EmptyResponse:
options.secho("Hmmm, no data returned. Try again with different "
"arguments.", fg="yellow")
options.say(Message.EMPTY_RESPONSE)
return
options.vecho("Caching data", nl=False, err=True)
options.say(Message.CACHE_WRITE)
with dots():
json_data = _projects_to_json(data)
write_json_cache(json_data, options.cache_key)
@ -116,7 +115,7 @@ async def _outdated(options: Options) -> None:
options.secho(bump.new_version, fg="green")
if len(outdated_set) == 0:
options.secho("Congrats! You have nothing to do!", fg="green")
options.say(Message.NO_WORK)
@click.command()