Examples¶
Some Clinner examples.
Simple Main¶
Example of a simple main with two defined commands foo and bar.
#!/usr/bin/env python
import os
import shlex
import sys
from clinner.command import command, Type as CommandType
from clinner.run.main import Main
@command(command_type=CommandType.SHELL
args=(('-i', '--input'),
('-o', '--output')),
parser_opts={'help': 'Foo command'})
def foo(*args, **kwargs):
"""List of foo commands"""
ls_cmd = shlex.split('ls')
wc_cmd = shlex.split('wc')
wc_cmd += [kwargs['input'], kwargs['output']]
return [ls_cmd, wc_cmd]
@command(command_type=CommandType.PYTHON,
parser_opts={'help': 'Bar command'})
def bar(*args, **kwargs):
"""Do a bar."""
return True
if __name__ == '__main__':
sys.exit(Main().run())
Builder Main¶
Example of main module with build utilities such as unit tests, lint, sphinx doc, tox and dist packaging:
#!/usr/bin/env python
import sys
from clinner.run import Main
class Build(Main):
commands = (
'clinner.run.commands.pytest.pytest',
'clinner.run.commands.prospector.prospector',
'clinner.run.commands.sphinx.sphinx',
'clinner.run.commands.tox.tox',
'clinner.run.commands.dist.dist',
)
if __name__ == '__main__':
sys.exit(Build().run())
Django Main¶
Example of main module for a Django application that uses uwsgi, health-check, prospector, pytest.
#!/usr/bin/env python3.6
"""Run script.
"""
import argparse
import multiprocessing
import os
import shlex
import sys
from socket import gethostname
from typing import List
import hvac
from clinner.command import Type as CommandType, command
from clinner.run import HealthCheckMixin, Main as BaseMain
from django.core.exceptions import ImproperlyConfigured
PYTHON = 'python3.6'
COVERAGE = 'coverage'
PROSPECTOR = 'prospector'
HEALTH_CHECK = 'health_check'
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
@command(command_type=CommandType.SHELL)
def migrate(*args, **kwargs) -> List[List[str]]:
cmd = shlex.split(f'{PYTHON} manage.py migrate')
cmd += args
return [cmd]
@command(command_type=CommandType.SHELL)
def build(*args, **kwargs) -> List[List[str]]:
return migrate('--fake-initial') + collectstatic('--noinput')
@command(command_type=CommandType.SHELL)
def manage(*args, **kwargs) -> List[List[str]]:
cmd = shlex.split(f'{PYTHON} manage.py')
cmd += args
return [cmd]
@command(command_type=CommandType.SHELL)
def unit_tests(*args, **kwargs) -> List[List[str]]:
parallel_count = multiprocessing.cpu_count()
coverage_erase = shlex.split(f'{COVERAGE} erase')
tests = shlex.split(f'{COVERAGE} run --concurrency=multiprocessing manage.py test --parallel {parallel_count}')
tests += args
coverage_combine = shlex.split(f'{COVERAGE} combine')
coverage_report = shlex.split(f'{COVERAGE} report')
coverage_xml = shlex.split(f'{COVERAGE} xml')
coverage_html = shlex.split(f'{COVERAGE} html')
return [coverage_erase, tests, coverage_combine, coverage_xml, coverage_html, coverage_report]
@command(command_type=CommandType.SHELL)
def prospector(*args, **kwargs) -> List[List[str]]:
cmd = [PROSPECTOR]
cmd += args
return [cmd]
@command(command_type=CommandType.SHELL)
def runserver(*args, **kwargs) -> List[List[str]]:
cmd = shlex.split(f'{PYTHON} manage.py runserver --nothreading')
cmd += args
return migrate('--fake-initial') + [cmd]
@command(command_type=CommandType.SHELL)
def uwsgi(*args, **kwargs) -> List[List[str]]:
http = f':{os.environ["APP_PORT"]}'
stats = f':{os.environ["STATS_PORT"]}'
ini = 'uwsgi.ini'
cmd = ['uwsgi', '--http', http, '--stats', stats, '--ini', ini]
cmd += args
return migrate('--fake-initial') + [cmd]
@command(command_type=CommandType.SHELL)
def collectstatic(*args, **kwargs) -> List[List[str]]:
cmd = shlex.split(f'{PYTHON} manage.py collectstatic')
cmd += args
return [cmd]
@command(command_type=CommandType.SHELL)
def shell(*args, **kwargs) -> List[List[str]]:
cmd = shlex.split(f'{PYTHON} manage.py shell')
cmd += args
return [cmd]
@command(command_type=CommandType.SHELL)
def health_check(*args, **kwargs) -> List[List[str]]:
"""
Run health-check
"""
cmd = [HEALTH_CHECK]
cmd += args
return [cmd]
class Main(HealthCheckMixin, BaseMain):
commands = (
'migrate',
'build',
'manage',
'unit_tests',
'prospector',
'runserver',
'uwsgi',
'collectstatic',
'shell',
'health_check',
)
def add_arguments(self, parser: argparse.ArgumentParser):
parser.add_argument('-s', '--settings', default='Development', help='Settings module')
def inject_app_settings(self):
"""
Injecting own settings.
"""
config_name = self.args.settings
os.environ['APP_HOST'] = os.environ.get('HOSTNAME', os.environ.get('APP_HOST', '0.0.0.0'))
os.environ['APP_PORT'] = os.environ.get('PORT_8000', os.environ.get('APP_PORT', '8000'))
os.environ['STATS_PORT'] = os.environ.get('PORT_9000', os.environ.get('STATS_PORT', '9000'))
# Django
os.environ['DJANGO_SETTINGS_MODULE'] = 'your_app.settings'
os.environ['DJANGO_CONFIGURATION'] = self.args.settings
# Plugins
os.environ['CLINNER_SETTINGS'] = self.args.settings or f'your_app.plugins_settings.clinner:{self.args.settings}'
os.environ['HEALTH_CHECK_SETTINGS'] = f'your_app.plugins_settings.health_check:{self.args.settings}'
def health_check(self):
"""
Does a check using Health Check application.
:return: 0 if healthy.
"""
return not self.run_command('manage', 'health_check', 'health', '-e')
if __name__ == '__main__':
sys.exit(Main().run())