Initial commit

.gitignore

@@ -0,0 +1,172 @@
+
+# Created by https://www.gitignore.io/api/macos,linux,python,windows,visualstudiocode
+
+### Linux ###
+*~
+
+# temporary files which can be created if a process still has a handle open of a deleted file
+.fuse_hidden*
+
+# KDE directory preferences
+.directory
+
+# Linux trash folder which might appear on any partition or disk
+.Trash-*
+
+# .nfs files are created when an open file is removed but is still being accessed
+.nfs*
+
+### macOS ###
+*.DS_Store
+.AppleDouble
+.LSOverride
+
+# Icon must end with two \r
+Icon
+
+# Thumbnails
+._*
+
+# Files that might appear in the root of a volume
+.DocumentRevisions-V100
+.fseventsd
+.Spotlight-V100
+.TemporaryItems
+.Trashes
+.VolumeIcon.icns
+.com.apple.timemachine.donotpresent
+
+# Directories potentially created on remote AFP share
+.AppleDB
+.AppleDesktop
+Network Trash Folder
+Temporary Items
+.apdisk
+
+### Python ###
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.coverage
+.coverage.*
+.cache
+.pytest_cache/
+nosetests.xml
+coverage.xml
+*.cover
+.hypothesis/
+
+# Translations
+*.mo
+*.pot
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# pyenv
+.python-version
+
+# celery beat schedule file
+celerybeat-schedule.*
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+
+### VisualStudioCode ###
+.vscode/*
+.history
+
+### Windows ###
+# Windows thumbnail cache files
+Thumbs.db
+ehthumbs.db
+ehthumbs_vista.db
+
+# Folder config file
+Desktop.ini
+
+# Recycle Bin used on file shares
+$RECYCLE.BIN/
+
+# Windows Installer files
+*.cab
+*.msi
+*.msm
+*.msp
+
+# Windows shortcuts
+*.lnk
+
+
+# End of https://www.gitignore.io/api/macos,linux,python,windows,visualstudiocode

api/me.py

@@ -0,0 +1,9 @@
+import requests
+from requests.auth import HTTPBasicAuth
+from typing import Dict, Any
+
+
+def get(username: str, password: str) -> Dict[str, Any]:
+    r = requests.get('https://www.pivotaltracker.com/services/v5/me', 
+                     auth=HTTPBasicAuth(username, password))
+    return r.json()

api/projects.py

@@ -0,0 +1,67 @@
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+import requests
+
+
+def get(token: str) -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects',
+        headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_project(token: str, project_id: int) -> Dict[str, Any]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}',
+        headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_stories(token: str, project_id: int) -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        '/stories', headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_memberships(token: str, project_id: int) -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        '/memberships', headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_iterations(token: str, project_id: int, scope: str) \
+        -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        f'/iterations?scope={scope}', headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_story_transitions(token: str, project_id: int,
+                          after: Optional[datetime] = None,
+                          before: Optional[datetime] = None) \
+        -> List[Dict[str, Any]]:
+    parameters = {
+        'occurred_after': after.isoformat() if after else None,
+        'occurred_before': before.isoformat() if before else None,
+    }
+
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        f'/story_transitions',
+        headers={'X-TrackerToken': token}, 
+        params=parameters)
+    return r.json()
+
+
+def get_history_days(token: str, project_id: int, start: datetime) \
+        -> Dict[str, Any]:
+    parameters = {'start_date': start.isoformat()}
+
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        f'/history/days', headers={'X-TrackerToken': token}, params=parameters)
+    return r.json()

api/stories.py

@@ -0,0 +1,33 @@
+import requests
+from typing import Dict, Any, List
+
+
+def get(token: str, story_id: int = None) -> Dict[str, Any]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/stories/{story_id}',
+        headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_tasks(token: str, project_id: int, story_id: int) \
+        -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        f'/stories/{story_id}/tasks', headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_comments(token: str, project_id: int, story_id: int) \
+        -> List[Dict[str, Any]]:
+    r = requests.get(
+        f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+        f'/stories/{story_id}/comments', headers={'X-TrackerToken': token})
+    return r.json()
+
+
+def get_blockers(token: str, project_id: int, story_id: int) \
+        -> List[Dict[str, Any]]:
+    r = requests.get(
+       f'https://www.pivotaltracker.com/services/v5/projects/{project_id}'
+       f'/stories/{story_id}/blockers', headers={'X-TrackerToken': token})
+    return r.json()

commands/__init__.py

@@ -0,0 +1,31 @@
+import api.projects
+from typing import Dict, Any
+from util import Color
+
+COLOR_TITLE = Color(Color.YELLOW)
+COLOR_HEADER = Color(Color.CYAN)
+COLOR_WHITE = Color(Color.BRIGHT_WHITE)
+
+
+def _format_state(state: str) -> str:
+    STATES = {
+        'accepted': '\033[92maccepted\033[0m',
+        'delivered': '\033[38;5;208mdelivered\033[0m',
+        'finished': '\033[94mfinished\033[0m',
+        'started': '\033[38;5;226mstarted\033[0m',
+        'planned': '\033[90mplanned\033[0m',
+        'unstarted': '\033[90munstarted\033[0m',
+    }
+
+    return STATES[state]
+
+
+def _get_persons(token: str, project_id: int) -> Dict[int, Dict[str, Any]]:
+    memberships = api.projects.get_memberships(token, project_id)
+
+    persons: Dict[int, Dict[str, Any]] = {}
+    for membership in memberships:
+        person = membership['person']
+        persons[person['id']] = person
+
+    return persons

commands/_stories_info.py

@@ -0,0 +1,151 @@
+import argparse
+import re
+import sys
+from datetime import datetime
+from typing import Any, Dict
+
+import base32_crockford
+
+import api.stories
+from config import Config
+from util import print_wrap
+
+from . import (COLOR_HEADER, COLOR_TITLE, COLOR_WHITE, _format_state,
+               _get_persons)
+
+
+def __print_story(story: Dict[str, Any], persons: Dict[int, Any]) -> None:
+    """
+    Prints the title, the current state and the estimate of the story, if
+    available.
+
+    TODO: Split up in functions.
+    """
+    COLOR_TITLE.print(story['name'], end='\n\n')
+
+    if 'current_state' in story:
+        state = _format_state(story['current_state'])
+        COLOR_HEADER.print('State:', state, end='\n\n')
+
+    if 'estimate' in story:
+        COLOR_HEADER.print('Estimate: ', end='')
+        print(story['estimate'], 'points', end='')
+        if len(story.get('owner_ids', [])) > 1:
+            points = story['estimate'] / len(story['owner_ids'])
+            print(f' ({points} each)', end='')
+        print(end='\n\n')
+
+
+def __print_owners(story: Dict[str, Any], persons: Dict[int, Any]) -> None:
+    """Prints the owners of the story, if available."""
+    if story.get('owner_ids'):
+        COLOR_HEADER.print('Owners:')
+
+        owners = []
+        for owner_id in story['owner_ids']:
+            name = persons[owner_id]['name']
+            initials = persons[owner_id]['initials']
+            owners.append(f'  - {name} ({initials})')
+
+        print('\n'.join(owners), end='\n\n')
+
+
+def __print_description(story: Dict[str, Any]) -> None:
+    """Prints the description of the story, if available."""
+    COLOR_HEADER.print('Description:')
+    if 'description' in story:
+        description = story['description'].strip()
+        print_wrap(description, indent='  ', end='\n\n')
+    else:
+        print('  (No description)', end='\n\n')
+
+
+def __print_labels(story: Dict[str, Any]) -> None:
+    """Prints the labels of the story, if available."""
+    if not story.get('labels'):
+        return
+
+    COLOR_HEADER.print('Labels:')
+
+    template = '\033[97;48;5;22m {} \033[0m'
+    labels = '  '.join(template.format(label['name'])
+                       for label in story['labels'])
+
+    print(f'  {labels}', end='\n\n')
+
+
+def __print_tasks(token: str, project_id: int, story_id: int) -> None:
+    """Prints the tasks of the story, if available."""
+    tasks = api.stories.get_tasks(token, project_id, story_id)
+
+    if tasks:
+        COLOR_HEADER.print('Tasks:')
+
+    for task in tasks:
+        print(end='  ')
+        print('[X]' if task['complete'] else '[ ]', end=' \033[34m')
+        print(base32_crockford.encode(task['id']), end=':\033[0m ')
+        print(task['description'])
+
+    print()
+
+
+def __print_comments(token: str, project_id: int, story_id: int,
+                     persons: Dict[int, Dict[str, Any]]) -> None:
+    """Prints the comments on the story, if available."""
+    comments = api.stories.get_comments(token, project_id, story_id)
+
+    if comments:
+        COLOR_HEADER.print('Comments:')
+
+    for comment in comments:
+        text = comment['text'].strip()
+        print_wrap(text, indent='  ')
+
+        person_id = comment['person_id']
+        name = persons[person_id]['name']
+        COLOR_WHITE.print('  -', name, end=' ')
+
+        date = datetime.strptime(comment['created_at'], '%Y-%m-%dT%H:%M:%SZ')
+        date_str = date.strftime('on %a %Y-%m-%d at %H:%M')
+        print(date_str, end='\n\n')
+
+
+def __print_blockers(token: str, project_id: int, story_id: int) -> None:
+    """Prints the stories that block this story, if available."""
+    blockers = api.stories.get_blockers(token, project_id, story_id)
+
+    if blockers:
+        COLOR_HEADER.print('Blockers:')
+
+    def blocker_repl(matchgroup: Any) -> str:
+        id = int(matchgroup.group(1))
+        code = base32_crockford.encode(id)
+        return COLOR_HEADER.format(code)
+
+    pattern = re.compile(r'#(\d+)')
+    for blocker in blockers:
+        resolved = 'X' if blocker['resolved'] else ' '
+        desc = pattern.sub(blocker_repl, blocker['description'])
+        print(f'  [{resolved}] {desc}')
+
+
+def _stories_info(args: argparse.Namespace) -> None:
+    try:
+        token = Config['user']['api_token']
+    except KeyError:
+        sys.exit(1)
+
+    story_id = base32_crockford.decode(args.story)
+    story = api.stories.get(token, story_id)
+
+    project_id = story['project_id']
+    persons = _get_persons(token, project_id)
+
+    __print_story(story, persons)
+    __print_owners(story, persons)
+    __print_description(story)
+    __print_labels(story)
+    __print_tasks(token, project_id, story_id)
+    __print_comments(token, project_id, story_id, persons)
+    __print_blockers(token, project_id, story_id)

commands/login.py

@@ -0,0 +1,16 @@
+import argparse
+import getpass
+
+import api.me
+from config import Config
+
+
+def login(arguments: argparse.Namespace) -> None:
+    username = input('E-mail: ')
+    password = getpass.getpass()
+
+    me = api.me.get(username, password)
+
+    Config['user']['api_token'] = me['api_token']
+    Config['user']['initials'] = me['initials']
+    Config.write()

commands/projects.py

@@ -0,0 +1,51 @@
+import argparse
+import sys
+from typing import Dict
+
+import base32_crockford
+import tabulate
+
+import api.projects
+from config import Config
+from util import require_login
+
+
+@require_login
+def list_projects(arguments: argparse.Namespace) -> None:
+    projects = api.projects.get(Config['user']['api_token'])
+    projects.sort(key=lambda project: project['name'])
+
+    aliases: Dict[int, str] = {}
+    for alias, alias_id in Config['project_aliases'].items():
+        aliases[int(alias_id)] = alias
+
+    table = []
+    for project in sorted(projects, key=lambda a: a['name']):
+        code = base32_crockford.encode(project['id'])
+        alias = aliases.get(project['id'])
+        table.append((code, project['name'], alias))
+
+    print(tabulate.tabulate(table, headers=('Code', 'Name', 'Alias')))
+
+
+def alias(arguments: argparse.Namespace) -> None:
+    project_id = base32_crockford.decode(arguments.code)
+    Config['project_aliases'][arguments.alias] = str(project_id)
+    Config.write()
+
+
+def rmalias(arguments: argparse.Namespace) -> None:
+    del Config['project_aliases'][arguments.alias]
+    Config.write()
+
+
+def info(arguments: argparse.Namespace) -> None:
+    try:
+        token = Config['user']['api_token']
+        project_id = int(Config['project_aliases'][arguments.alias])
+    except KeyError:
+        print(f'unknown alias {arguments.alias}')
+        sys.exit(1)
+
+    projects = api.projects.get_project(token, project_id).items()
+    print(tabulate.tabulate(projects))

commands/stories.py

@@ -0,0 +1,147 @@
+import argparse
+from beeprint import pp
+from collections import defaultdict
+from datetime import datetime, timedelta
+from typing import Any, DefaultDict, Dict, List, Sequence, Tuple
+
+import base32_crockford
+import tabulate
+
+import api.projects
+import api.stories
+from config import Config
+from util import require_login
+
+from . import COLOR_HEADER, _format_state, _get_persons
+from ._stories_info import _stories_info
+
+STATES = 'unstarted', 'planned', 'started', 'finished', 'delivered', 'accepted'
+
+
+Persons = Dict[int, Dict[str, Any]]
+Totals = DefaultDict[int, Dict[str, int]]
+
+
+def __get_row(item: Tuple[int, Dict[str, int]], persons: Persons) -> Sequence:
+    owner_id, points = item
+    name = persons[owner_id]['name']
+
+    estimates = [points[state] for state in STATES]
+
+    progress = '['
+    progress += f'\033[90m' + 'P' * round(points['planned'])
+    progress += f'\033[38;5;226m' + 'S' * round(points['started'])
+    progress += f'\033[94m' + 'F' * round(points['finished'])
+    progress += f'\033[38;5;208m' + 'D' * round(points['delivered'])
+    progress += f'\033[92m' + 'A' * round(points['accepted'])
+    progress += '\033[0m]'
+
+    return (name, *estimates, sum(estimates), progress)
+
+
+def __format_story(story: Dict[str, Any], persons: Persons, totals: Totals) \
+        -> Tuple[str, str, str, str, str, int]:
+    code = base32_crockford.encode(story['id'])
+
+    is_owner = False
+    initials = []
+    owner_ids = story.get('owner_ids', [])
+    for owner_id in owner_ids:
+        value = persons[owner_id]['initials']
+        if value == Config['user']['initials']:
+            is_owner = True
+            value = f'\033[96m{value}\033[97m'
+        initials.append(value)
+
+    estimate = story.get('estimate')
+    if estimate:
+        for owner_id in owner_ids:
+            state = story['current_state']
+            totals[owner_id][state] += estimate / len(owner_ids)
+
+    if is_owner:
+        code = f'\033[1;97m{code}'
+        estimate = f'\033[97m{estimate}\033[0m' if estimate else None
+
+    owners = ', '.join(initials)
+
+    state = _format_state(story['current_state'])
+    return code, story['story_type'], story['name'], owners, state, estimate
+
+
+def __print_stories(stories: List[Dict[str, Any]], persons: Persons,
+                    totals: Totals) -> None:
+    table = [__format_story(story, persons, totals) for story in stories]
+    headers = 'Code', 'Type', 'Story name', 'Owners', 'State', 'Pts'
+    print(tabulate.tabulate(table, headers=headers), end='\n\n')
+
+
+def __print_totals(totals: Totals, persons: Persons) -> None:
+    COLOR_HEADER.print('Point totals:', end='\n\n')
+
+    state_headers = [_format_state(state) for state in STATES]
+    headers = ('Owner', *state_headers, 'Total', 'Progress')
+
+    data = [__get_row(item, persons) for item in totals.items()]
+    data.sort(key=lambda row: row[-2])
+    print(tabulate.tabulate(data, headers), end='\n\n')
+
+
+def __print_burndown(token: str, iteration: Dict[str, Any], persons: Persons,
+                     hide_accepted: bool = False) -> None:
+    COLOR_HEADER.print('Burndown:', end='\n\n')
+
+    start = datetime.strptime(iteration['start'], '%Y-%m-%dT%H:%M:%SZ')
+    history = api.projects.get_history_days(
+        token, iteration['project_id'], start - timedelta(days=1))
+
+    accepted_points = history['data'][0][1]
+
+    for date, *counts in history['data'][1:]:
+        progress = ''
+
+        if not hide_accepted:
+            progress += '\033[92m' + 'A' * round(counts[0] - accepted_points)
+
+        progress += '\033[38;5;208m' + 'D' * round(counts[1])
+        progress += '\033[94m' + 'F' * round(counts[2])
+        progress += '\033[38;5;226m' + 'S' * round(counts[3])
+        progress += '\033[90m' + 'P' * round(counts[5])
+        progress += '\033[0m'
+
+        print(f'{date}: {progress}')
+
+    print()
+
+
+def _stories_current(arguments: argparse.Namespace) -> None:
+    try:
+        project_id = int(Config['project_aliases'][arguments.project])
+    except KeyError:
+        project_id = base32_crockford.decode(arguments.project)
+
+    token = Config['user']['api_token']
+
+    iterations = api.projects.get_iterations(
+        token, project_id, scope=arguments.scope)
+    if not iterations:
+        print('No current iteration.')
+        return
+
+    persons = _get_persons(token, project_id=project_id)
+    totals: DefaultDict[int, Dict[str, int]] = \
+        defaultdict(lambda: dict((state, 0) for state in STATES))
+
+    iteration = iterations[0]
+
+    __print_stories(iteration['stories'], persons, totals)
+    __print_totals(totals, persons)
+    __print_burndown(token, iteration, persons, arguments.hide_accepted)
+
+
+@require_login
+def stories(arguments: argparse.Namespace) -> None:
+    if arguments.story:
+        _stories_info(arguments)
+    else:
+        _stories_current(arguments)

config.py

@@ -0,0 +1,32 @@
+import configparser
+import os
+from typing import Dict
+
+import appdirs
+
+
+class _Config(type):
+    def __getitem__(cls, key: str) -> Dict[str, str]:
+        if key not in cls.config:
+            cls.config[key] = {}
+
+        return cls.config[key]
+
+
+class Config(object, metaclass=_Config):
+    config = configparser.ConfigParser()
+
+    @staticmethod
+    def __get_filename() -> str:
+        data_dir = appdirs.user_data_dir('pivotalcli', 'sijmen')
+        os.makedirs(data_dir, exist_ok=True)
+        return os.path.join(data_dir, 'config.ini')
+
+    @classmethod
+    def read(cls) -> None:
+        cls.config.read(cls.__get_filename())
+
+    @classmethod
+    def write(cls) -> None:
+        with open(cls.__get_filename(), mode='w') as config_file:
+            cls.config.write(config_file)

pivotalcli.py

@@ -0,0 +1,52 @@
+#!/usr/bin/env python3
+import argparse
+import commands.login as cmd_login
+import commands.projects as cmd_projects
+import commands.stories as cmd_stories
+
+from config import Config
+
+
+def parse_arguments() -> None:
+    parser = argparse.ArgumentParser()
+    parser.set_defaults(func=lambda _: parser.print_help())
+    commands = parser.add_subparsers(title='commands')
+
+    login_parser = commands.add_parser('login')
+    login_parser.set_defaults(func=cmd_login.login)
+
+    projects_parser = commands.add_parser('projects')
+    projects_parser.set_defaults(func=cmd_projects.list_projects)
+
+    projects_commands = projects_parser.add_subparsers(title='commands')
+    projects_list_parser = projects_commands.add_parser('list')
+    projects_list_parser.set_defaults(func=cmd_projects.list_projects)
+
+    projects_alias_parser = projects_commands.add_parser('alias')
+    projects_alias_parser.add_argument('code', type=str)
+    projects_alias_parser.add_argument('alias', type=str)
+    projects_alias_parser.set_defaults(func=cmd_projects.alias)
+
+    projects_rmalias_parser = projects_commands.add_parser('rmalias')
+    projects_rmalias_parser.add_argument('alias', type=str)
+    projects_rmalias_parser.set_defaults(func=cmd_projects.rmalias)
+
+    projects_info_parser = projects_commands.add_parser('info')
+    projects_info_parser.add_argument('alias', type=str)
+    projects_info_parser.set_defaults(func=cmd_projects.info)
+
+    stories_parser = commands.add_parser('stories', description='story stuff')
+    stories_parser.set_defaults(func=cmd_stories.stories)
+    stories_parser.add_argument('project', type=str)
+    stories_parser.add_argument('story', type=str, nargs='?', default=None)
+    stories_parser.add_argument('--scope', type=str, default='current')
+    stories_parser.add_argument('--hide-accepted', nargs='?', type=bool,
+                                const=True, default=False)
+
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == '__main__':
+    Config.read()
+    parse_arguments()

requirements.txt

@@ -0,0 +1,37 @@
+appnope==0.1.0
+astroid==1.6.1
+autopep8==1.3.4
+base32-crockford==0.3.0
+beeprint==2.4.7
+certifi==2018.1.18
+chardet==3.0.4
+decorator==4.2.1
+flake8==3.5.0
+idna==2.6
+ipython==6.2.1
+ipython-genutils==0.2.0
+isort==4.3.4
+jedi==0.11.1
+lazy-object-proxy==1.3.1
+mccabe==0.6.1
+mypy==0.570
+parso==0.1.1
+pexpect==4.4.0
+pickleshare==0.7.4
+prompt-toolkit==1.0.15
+ptyprocess==0.5.2
+pycodestyle==2.3.1
+pyflakes==1.6.0
+Pygments==2.2.0
+pylint==1.8.2
+requests==2.18.4
+rope==0.10.7
+simplegeneric==0.8.1
+six==1.11.0
+tabulate==0.8.2
+traitlets==4.3.2
+typed-ast==1.1.0
+urllib3==1.22
+urwid==2.0.1
+wcwidth==0.1.7
+wrapt==1.10.11

util.py

@@ -0,0 +1,77 @@
+import commands.login
+import shutil
+import textwrap
+from typing import Any, Callable, Optional
+
+from config import Config
+
+
+class Color:
+    BLACK = 30
+    RED = 31
+    GREEN = 32
+    YELLOW = 33
+    BLUE = 34
+    MAGENTA = 35
+    CYAN = 36
+    WHITE = 37
+    BRIGHT_BLACK = 90
+    BRIGHT_RED = 91
+    BRIGHT_GREEN = 92
+    BRIGHT_YELLOW = 93
+    BRIGHT_BLUE = 94
+    BRIGHT_MAGENTA = 95
+    BRIGHT_CYAN = 96
+    BRIGHT_WHITE = 97
+
+    RESET = 0
+
+    def __init__(self, foreground: Optional[int],
+                 background: Optional[int] = None) -> None:
+        self.foreground = foreground
+        self.background = background
+
+    def print(self, *args: Any, end: str = '\n') -> None:
+        print(self.format(*args), end=end)
+
+    def format(self, *args: Any) -> str:
+        colors = ['0']
+
+        if self.foreground is not None:
+            colors.append(str(self.foreground))
+
+        if self.background is not None:
+            colors.append(str(self.background + 10))
+
+        color_str = ';'.join(colors)
+        text = ' '.join(str(a) for a in args)
+
+        return f'\033[{color_str}m{text}\033[0m'
+
+
+def print_wrap(text: str, indent: str = '', end: str = '\n') -> None:
+    w, _ = shutil.get_terminal_size((80, 20))
+    if w > 72:
+        w = 72
+
+    lines = text.split('\n')
+    for i, line in enumerate(lines):
+        _end = '\n' if i < len(lines) - 1 else end
+
+        if not line:
+            print(indent, end=_end)
+            continue
+
+        wrapped = textwrap.fill(
+            line, w, initial_indent=indent, subsequent_indent=indent)
+        print(wrapped, end=_end)
+
+
+def require_login(function: Callable) -> Callable:
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        while 'api_token' not in Config['user']:
+            commands.login.login(None)
+
+        return function(*args, **kwargs)
+
+    return wrapper