check_for_updates.py 3.3 KB
Newer Older
Stefan Scherfke's avatar
Stefan Scherfke committed
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121
"""
Check if there are updates available for external dependencies.

.. code-block:: console

   $ ownconda check-for-updates ../external-recipes/

By default, only packages that are not up-to-date are shown.  Use the
``--show-uptodate`` option to display all packages.

"""
from functools import partial as P

import click
import trio

from .. import click_util, recipes, update_check


@click.command()
@click_util.argument.recipe_root()
@click.option(
    '--hide-uptodate/--show-uptodate',
    '-h/-s',
    default=True,
    show_default=True,
    help='Hide packages that are up-to-date',
)
@click.option(
    '--verbose',
    '-v',
    is_flag=True,
    default=False,
    help='Show changelog URLs for each package')
def cli(recipe_root, hide_uptodate, verbose):
    """Update check for external packages in RECIPE_ROOT.

    You can specify multiple recipe roots which are all searched recursively
    for recipes.

    """
    trio.run(check_for_updates, recipe_root, hide_uptodate, verbose)


async def check_for_updates(recipe_root, hide_uptodate, verbose):
    """Asynchronously run the ckeckers of all *recipes* and print the results
    to *stdout*.

    *recipies* is a list as returned by :func:`get_recipes()`.

    """
    # Create tasks and run them asyncronously
    recipe_data = list(recipes.load_recipes(recipe_root))
    tasks = [P(update_check.run_check, meta, path) for meta, path in recipe_data]
    results = await run_with_progress(tasks)

    # All tasks have finished and we can print the results
    click_util.info('Package: latest version (current version)')
    for (meta, _path), result in zip(recipe_data, results):
        if isinstance(result, Exception):
            click_util.error(f'{result.recipe}: ERROR: {result.msg}')
            continue

        name, latest, current = result
        if hide_uptodate and latest == current:
            continue

        # Format results
        msg = f'{name} {latest} ({current})'
        color = _get_color(latest, current)

        if verbose:
            try:
                changelog_url = meta['extra']['changelog_url']

                current = str(current)
                latest = str(latest)
                changelog_url = changelog_url.replace(current, latest)

                c_minor = '.'.join(current.split('.')[:2])
                l_minor = '.'.join(latest.split('.')[:2])
                changelog_url = changelog_url.replace(c_minor, l_minor)

                msg = f'{msg}:\n  {changelog_url}\n'
            except KeyError:
                msg = f'{msg}\n'

        click_util.echo(msg, fg=color)


async def run_with_progress(tasks):
    results = [None] * len(tasks)

    with click.progressbar(length=len(tasks), fill_char='█') as bar:
        async with trio.open_nursery() as nursery:
            for key, t in enumerate(tasks):
                nursery.start_soon(fetch, results, key, bar, t)

    return results


async def fetch(results, key, bar, task_func):
    try:
        result = await task_func()
        results[key] = result
    except update_check.UpdateCheckFailed as e:
        results[key] = e
    bar.update(1)


def _get_color(latest, current):
    if latest is None:
        color = 'yellow'
    elif latest == current:
        color = 'green'
    elif latest > current:
        color = 'red'
    else:
        color = 'magenta'

    return color