Vendor colorclass

.gitattributes

@@ -1,3 +1,4 @@
 CHANGES.rst merge=union
 docs/* linguist-documentation
 indico/translations/*.pot linguist-generated
+indico/vendor/* linguist-vendored

bin/maintenance/update_backrefs.py

@@ -13,7 +13,6 @@
 from sqlalchemy import inspect
 
 from indico.core.db.sqlalchemy.util.models import get_all_models, import_all_models
-from indico.util.console import cformat
 
 
 def _find_backrefs():
@@ -76,10 +75,10 @@ def main(ci):
         if in_backrefs and not backrefs_written:
             _write_backrefs(rels, new_source)
         if not backrefs_written:
-            print(cformat('%{yellow}Class {} has no comment for backref information').format(cls.__name__))
+            click.secho(f'Class {cls.__name__} has no comment for backref information', fg='yellow')
             has_missing = True
         if source != new_source:
-            print(cformat('%{green!}Updating backref info for {} in {}').format(cls.__name__, path))
+            click.secho(f'Updating backref info for {cls.__name__} in {path}', fg='green', bold=True)
             has_updates = True
             with open(path, 'w') as f:
                 f.writelines(line + '\n' for line in new_source)

bin/maintenance/update_header.py

@@ -198,15 +198,16 @@ def main(ctx, ci, year, path):
                     if update_header(filepath, year, ci):
                         error = True
         except subprocess.CalledProcessError:
-            raise click.UsageError(cformat('%{red!}You must be within a git repository to run this script.'))
+            raise click.UsageError(
+                click.style('You must be within a git repository to run this script.', fg='red', bold=True))
 
     if not error:
-        print(cformat('%{green}\u2705 All headers are up to date'))
+        click.secho('\u2705 All headers are up to date', fg='green')
     elif ci:
-        print(cformat('%{red}\u274C Some headers need to be updated or added'))
+        click.secho('\u274C Some headers need to be updated or added', fg='red')
         sys.exit(1)
     else:
-        print(cformat('%{yellow}\U0001F504 Some headers have been updated (or are missing)'))
+        click.secho('\U0001F504 Some headers have been updated (or are missing)', fg='yellow')
 
 
 if __name__ == '__main__':

indico/cli/database.py

@@ -115,7 +115,7 @@ def _safe_downgrade(*args, **kwargs):
                       '%{red!}Debug mode is NOT ACTIVE, so make sure you are on the right machine!'))
     if not skip_confirm and input(cformat('%{yellow!}***%{reset} '
                                           'To confirm this, enter %{yellow!}YES%{reset}: ')) != 'YES':
-        print(cformat('%{green}Aborted%{reset}'))
+        click.secho('Aborted', fg='green')
         sys.exit(1)
     else:
         return func(*args, **kwargs)
@@ -140,9 +140,9 @@ def _call_with_plugins(*args, **kwargs):
         alembic.command.ScriptDirectory = PluginScriptDirectory
         for plugin in plugins:
             if not os.path.exists(plugin.alembic_versions_path):
-                print(cformat("%{cyan}skipping plugin '{}' (no migrations folder)").format(plugin.name))
+                click.secho(f"skipping plugin '{plugin.name}' (no migrations folder)", fg='cyan')
                 continue
-            print(cformat("%{cyan!}executing command for plugin '{}'").format(plugin.name))
+            click.secho(f"executing command for plugin '{plugin.name}'", fg='cyan', bold=True)
             with plugin.plugin_context():
                 func(*args, **kwargs)
 

indico/cli/user.py

@@ -80,7 +80,7 @@ def search(substring, include_deleted, include_pending, include_blocked, include
                        include_blocked=include_blocked, external=include_external,
                        allow_system_user=include_system, **criteria)
     if not res:
-        print(cformat('%{yellow}No results found'))
+        click.secho('No results found', fg='yellow')
         return
     elif len(res) > 100:
         click.confirm(f'{len(res)} results found. Show them anyway?', abort=True)
@@ -120,7 +120,7 @@ def create(grant_admin):
         email = email.lower()
         if not User.query.filter(User.all_emails == email, ~User.is_deleted, ~User.is_pending).has_rows():
             break
-        print(cformat('%{red}Email already exists'))
+        click.secho('Email already exists', fg='red')
     first_name = click.prompt('First name').strip()
     last_name = click.prompt('Last name').strip()
     affiliation = click.prompt('Affiliation', '').strip()
@@ -129,7 +129,7 @@ def create(grant_admin):
         username = click.prompt('Enter username').lower().strip()
         if not Identity.query.filter_by(provider='indico', identifier=username).has_rows():
             break
-        print(cformat('%{red}Username already exists'))
+        click.secho('Username already exists', fg='red')
     password = prompt_pass()
     if password is None:
         return
@@ -139,7 +139,7 @@ def create(grant_admin):
     user.is_admin = grant_admin
     _print_user_info(user)
 
-    if click.confirm(cformat('%{yellow}Create the new {}?').format(user_type), default=True):
+    if click.confirm(click.style(f'Create the new {user_type}?', fg='yellow'), default=True):
         db.session.add(user)
         db.session.commit()
         print(cformat('%{green}New {} created successfully with ID: %{green!}{}').format(user_type, user.id))
@@ -151,16 +151,16 @@ def grant_admin(user_id):
     """Grant administration rights to a given user."""
     user = User.get(user_id)
     if user is None:
-        print(cformat('%{red}This user does not exist'))
+        click.secho('This user does not exist', fg='red')
         return
     _print_user_info(user)
     if user.is_admin:
-        print(cformat('%{yellow}This user already has administration rights'))
+        click.secho('This user already has administration rights', fg='yellow')
         return
-    if click.confirm(cformat('%{yellow}Grant administration rights to this user?')):
+    if click.confirm(click.style('Grant administration rights to this user?', fg='yellow')):
         user.is_admin = True
         db.session.commit()
-        print(cformat('%{green}Administration rights granted successfully'))
+        click.secho('Administration rights granted successfully', fg='green')
 
 
 @cli.command()
@@ -169,16 +169,16 @@ def revoke_admin(user_id):
     """Revoke administration rights from a given user."""
     user = User.get(user_id)
     if user is None:
-        print(cformat('%{red}This user does not exist'))
+        click.secho('This user does not exist', fg='red')
         return
     _print_user_info(user)
     if not user.is_admin:
-        print(cformat('%{yellow}This user does not have administration rights'))
+        click.secho('This user does not have administration rights', fg='yellow')
         return
-    if click.confirm(cformat('%{yellow}Revoke administration rights from this user?')):
+    if click.confirm(click.style('Revoke administration rights from this user?', fg='yellow')):
         user.is_admin = False
         db.session.commit()
-        print(cformat('%{green}Administration rights revoked successfully'))
+        click.secho('Administration rights revoked successfully', fg='green')
 
 
 @cli.command()
@@ -187,16 +187,16 @@ def block(user_id):
     """Block a given user."""
     user = User.get(user_id)
     if user is None:
-        print(cformat('%{red}This user does not exist'))
+        click.secho('This user does not exist', fg='red')
         return
     _print_user_info(user)
     if user.is_blocked:
-        print(cformat('%{yellow}This user is already blocked'))
+        click.secho('This user is already blocked', fg='yellow')
         return
-    if click.confirm(cformat('%{yellow}Block this user?')):
+    if click.confirm(click.style('Block this user?', fg='yellow')):
         user.is_blocked = True
         db.session.commit()
-        print(cformat('%{green}Successfully blocked user'))
+        click.secho('Successfully blocked user', fg='green')
 
 
 @cli.command()
@@ -205,16 +205,16 @@ def unblock(user_id):
     """Unblock a given user."""
     user = User.get(user_id)
     if user is None:
-        print(cformat('%{red}This user does not exist'))
+        click.secho('This user does not exist', fg='red')
         return
     _print_user_info(user)
     if not user.is_blocked:
-        print(cformat('%{yellow}This user is not blocked'))
+        click.secho('This user is not blocked', fg='yellow')
         return
-    if click.confirm(cformat('%{yellow}Unblock this user?')):
+    if click.confirm(click.style('Unblock this user?', fg='yellow')):
         user.is_blocked = False
         db.session.commit()
-        print(cformat('%{green}Successfully unblocked user'))
+        click.secho('Successfully unblocked user', fg='green')
 
 
 @cli.group('token')
@@ -229,7 +229,7 @@ def token_list(user_id, verbose):
     """List the tokens of the user."""
     user = User.get(user_id)
     if user is None:
-        click.echo(cformat('%{red!}This user does not exist'))
+        click.secho('This user does not exist', fg='red', bold=True)
         return
     _print_user_info(user)
     query = (
@@ -242,7 +242,7 @@ def token_list(user_id, verbose):
 
     tokens = query.all()
     if not tokens:
-        click.echo(cformat('%{yellow}This user has no tokens'))
+        click.secho('This user has no tokens', fg='yellow')
         return
 
     verbose_cols = ('Last IP', 'Use count', 'Status', 'ID') if verbose else ()
@@ -285,11 +285,11 @@ def token_create(user_id, token_name, scopes):
     """Create a personal token for a user."""
     user = User.get(user_id)
     if user is None:
-        click.echo(cformat('%{red!}This user does not exist'))
+        click.secho('This user does not exist', fg='red', bold=True)
         return
     token = user.query_personal_tokens().filter(db.func.lower(PersonalToken.name) == token_name.lower()).first()
     if token:
-        click.echo(cformat('%{red!}A token with this name already exists'))
+        click.secho('A token with this name already exists', fg='red', bold=True)
         return
     token = PersonalToken(user=user, name=token_name, scopes=scopes)
     access_token = token.generate_token()
@@ -316,11 +316,11 @@ def token_update(user_id, token_name, add_scopes, del_scopes, new_token_name, re
         return
     user = User.get(user_id)
     if user is None:
-        click.echo(cformat('%{red!}This user does not exist'))
+        click.secho('This user does not exist', fg='red', bold=True)
         return
     token = user.query_personal_tokens().filter(db.func.lower(PersonalToken.name) == token_name.lower()).first()
     if not token:
-        click.echo(cformat('%{red!}This token does not exist or has been revoked'))
+        click.secho('This token does not exist or has been revoked', fg='red', bold=True)
         return
     old_scopes = set(token.scopes)
     old_name = token.name
@@ -337,17 +337,17 @@ def token_update(user_id, token_name, add_scopes, del_scopes, new_token_name, re
             .has_rows()
         )
         if conflict:
-            click.echo(cformat('%{red!}A token with this name already exists'))
+            click.secho('A token with this name already exists', fg='red', bold=True)
             return
         token.name = new_token_name
     if reset_token:
         new_access_token = token.generate_token()
     db.session.commit()
-    click.echo(cformat("%{green}Token '{}' updated").format(old_name))
+    click.secho(f"Token '{old_name}' updated", fg='green')
     if token.name != old_name:
-        click.echo(cformat('Name: %{white!}{}').format(token.name))
+        click.secho(f'Name: {token.name}', bold=True)
     if new_access_token:
-        click.echo(cformat('Token: %{white!}{}').format(new_access_token))
+        click.secho(f'Token: {new_access_token}', bold=True)
     if token.scopes != old_scopes:
         click.echo(f'Scopes: {token.get_scope() or "-"}')
 
@@ -359,15 +359,15 @@ def token_revoke(user_id, token_name):
     """Revoke a user's personal token."""
     user = User.get(user_id)
     if user is None:
-        click.echo(cformat('%{red!}This user does not exist'))
+        click.secho('This user does not exist', fg='red', bold=True)
         return
     token = user.query_personal_tokens().filter(db.func.lower(PersonalToken.name) == token_name.lower()).first()
     if not token:
-        click.echo(cformat('%{red!}This token does not exist or has been revoked'))
+        click.secho('This token does not exist or has been revoked', fg='red', bold=True)
         return
     elif token.revoked_dt:
-        click.echo(cformat('%{yellow}This token is already revoked'))
+        click.secho('This token is already revoked', fg='yellow')
         return
     token.revoke()
     db.session.commit()
-    click.echo(cformat("%{green}Token '{}' revoked").format(token.name))
+    click.secho(f"Token '{token.name}' revoked", fg='green')

indico/cli/watchman.py

@@ -11,6 +11,7 @@
 import sys
 import time
 
+import click
 import pywatchman
 from flask.helpers import get_root_path
 from werkzeug._reloader import _find_watchdog_paths
@@ -62,7 +63,7 @@ def consume(self, client):
             files = sorted(f for record in data for f in record.get('files', []))
             relpath = os.path.relpath(self.path)
             if relpath == '.':
-                print(cformat('%{cyan}Changes found:').format(relpath))
+                click.secho('Changes found:', fg='cyan')
             else:
                 print(cformat('%{cyan}Changes found in %{cyan!}{}%{reset}%{cyan}:').format(relpath))
             for f in files:
@@ -144,26 +145,26 @@ def _monitor(self):
     def _launch(self, quiet=False, retry=0):
         assert not self._proc
         if not quiet and not retry:
-            print(cformat('%{green!}Launching Indico'))
+            click.secho('Launching Indico', fg='green', bold=True)
         try:
             argv = _disable_reloader(sys.argv)
             self._proc = subprocess.Popen(argv)
         except OSError as exc:
             delay = (retry + 1) * 0.5
-            print(cformat('%{red!}Could not launch Indico: {}').format(exc))
-            print(cformat('%{yellow}Retrying in {}s').format(delay))
+            click.secho(f'Could not launch Indico: {exc}', fg='red', bold=True)
+            click.secho(f'Retrying in {delay}s', fg='yellow')
             time.sleep(delay)
             self._launch(quiet=quiet, retry=(retry + 1))
 
     def _terminate(self, quiet=False):
         if not self._proc:
             return
         if not quiet:
-            print(cformat('%{red!}Terminating Indico'))
+            click.secho('Terminating Indico', fg='red', bold=True)
         self._proc.terminate()
         self._proc = None
 
     def _restart(self):
-        print(cformat('%{yellow!}Restarting Indico'))
+        click.secho('Restarting Indico', fg='yellow', bold=True)
         self._terminate(quiet=True)
         self._launch(quiet=True)

indico/core/celery/cli.py

@@ -9,7 +9,6 @@
 from celery.bin.celery import celery as celery_cmd
 
 from indico.core.celery.util import unlock_task
-from indico.util.console import cformat
 
 
 # remove the celery shell command
@@ -30,6 +29,6 @@ def unlock(name):
     """
 
     if unlock_task(name):
-        print(cformat('%{green!}Task {} unlocked').format(name))
+        click.secho(f'Task {name} unlocked', fg='green', bold=True)
     else:
-        print(cformat('%{yellow}Task {} is not locked').format(name))
+        click.secho(f'Task {name} is not locked', fg='yellow')