stakkr.cli.exec_cmd() - Code Metrics - Inspection of "Create .coveragerc" - stakkr-org/stakkr - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 61e6a6...28b40d )

by Emmanuel

created 2019-06-22 07:09 UTC

stakkr.cli.exec_cmd() A

↳ Parent: stakkr.cli

Complexity

Conditions

Size

Total Lines	20
Code Lines	13

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	6

Importance

Changes

Metric	Value
cc	2
eloc	13
nop	5
dl	0
loc	20
ccs	0
cts	12
cp	0
crap	6
rs	9.75
c	0
b	0
f	0

#!/usr/bin/env python
# coding: utf-8
"""
CLI Entry Point.

From click, build stakkr.
Give all options to manage services to be launched, stopped, etc.
"""

import sys
import click
from click.core import Context
from stakkr.docker_actions import get_running_containers_names
from stakkr.aliases import get_aliases


@click.group(help="""Main CLI Tool that easily create / maintain
a stack of services, for example for web development.

Read the configuration file and setup the required services by
linking and managing everything for you.""")
@click.version_option('4.0.3')
@click.option('--config', '-c', help='Set the configuration filename (stakkr.yml by default)')
@click.option('--debug/--no-debug', '-d', default=False)
@click.option('--verbose', '-v', is_flag=True)
@click.pass_context
def stakkr(ctx: Context, config=None, debug=False, verbose=True):
    """Click group, set context and main object."""
    from stakkr.actions import StakkrActions

    ctx.obj['CONFIG'] = config
    ctx.obj['DEBUG'] = debug
    ctx.obj['VERBOSE'] = verbose
    ctx.obj['STAKKR'] = StakkrActions(ctx.obj)


@stakkr.command(help="""Enter a container to perform direct actions such as
install packages, run commands, etc.""")
@click.argument('container', required=True)
@click.option('--user', '-u', help="User's name. Valid choices : www-data or root")
@click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
@click.pass_context
def console(ctx: Context, container: str, user: str, tty: bool):
    """See command Help."""
    ctx.obj['STAKKR'].init_project()
    ctx.obj['CTS'] = get_running_containers_names(ctx.obj['STAKKR'].project_name)
    if ctx.obj['CTS']:
        ct_choice = click.Choice(ctx.obj['CTS'])
        ct_choice.convert(container, None, ctx)

    ctx.obj['STAKKR'].console(container, _get_cmd_user(user, container), tty)


@stakkr.command(help="""Execute a command into a container.

Examples:\n
- ``stakkr -v exec mysql mysqldump -p'$MYSQL_ROOT_PASSWORD' mydb > /tmp/backup.sql``\n
- ``stakkr exec php php -v`` : Execute the php binary in the php container with option -v\n
- ``stakkr exec apache service apache2 restart``\n
""", name='exec', context_settings=dict(ignore_unknown_options=True, allow_interspersed_args=False))
@click.pass_context
@click.option('--user', '-u', help="User's name. Be careful, each container have its own users.")
@click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
@click.argument('container', required=True)
@click.argument('command', required=True, nargs=-1, type=click.UNPROCESSED)
def exec_cmd(ctx: Context, user: str, container: str, command: tuple, tty: bool):
    """See command Help."""
    ctx.obj['STAKKR'].init_project()
    ctx.obj['CTS'] = get_running_containers_names(ctx.obj['STAKKR'].project_name)
    if ctx.obj['CTS']:
        click.Choice(ctx.obj['CTS']).convert(container, None, ctx)

    ctx.obj['STAKKR'].exec_cmd(container, _get_cmd_user(user, container), command, tty)


@stakkr.command(help="Restart all (or a single as CONTAINER) container(s)")
@click.argument('container', required=False)
@click.option('--pull', '-p', help="Force a pull of the latest images versions", is_flag=True)
@click.option('--recreate', '-r', help="Recreate all containers", is_flag=True)
@click.option('--proxy/--no-proxy', '-P', help="Restart the proxy", default=True)
@click.pass_context
def restart(ctx: Context, container: str, pull: bool, recreate: bool, proxy: bool):
    """See command Help."""
    print(click.style('[RESTARTING]', fg='green') + ' your stakkr services')
    try:
        ctx.invoke(stop, container=container, proxy=proxy)
    except Exception:
        pass

    ctx.invoke(start, container=container, pull=pull, recreate=recreate, proxy=proxy)


@stakkr.command(help="""List available services available for stakkr.yml
(with info if the service is enabled)""")
@click.pass_context
def services(ctx):
    """See command Help."""
    ctx.obj['STAKKR'].init_project()

    from stakkr.stakkr_compose import get_available_services

    print('Available services usable in stakkr.yml ', end='')
    print('({} = disabled) : '.format(click.style('✘', fg='red')))

    svcs = ctx.obj['STAKKR'].get_config()['services']
    enabled_svcs = [svc for svc, opts in svcs.items() if opts['enabled'] is True]
    available_svcs = get_available_services(ctx.obj['STAKKR'].project_dir)
    for available_svc in sorted(list(available_svcs.keys())):
        sign = click.style('✘', fg='red')
        if available_svc in enabled_svcs:
            version = svcs[available_svc]['version']
            sign = click.style(str(version), fg='green')

        print('  - {} ({})'.format(available_svc, sign))


@stakkr.command(help="""Download a pack of services from github (see github) containing services.
PACKAGE is the git url or package name.
NAME is the directory name to clone the repo.
""")
@click.argument('package', required=True)
@click.argument('name', required=False)
@click.pass_context
def services_add(ctx: Context, package: str, name: str):
    """See command Help."""
    from stakkr.services import install

    project_dir = _get_project_dir(ctx.obj['CONFIG'])
    services_dir = '{}/services'.format(project_dir)
    name = package if name is None else name
    success, message = install(services_dir, package, name)
    if success is False:
        click.echo(click.style(message, fg='red'))
        sys.exit(1)

    stdout_msg = 'Package "' + click.style(package, fg='green') + '" installed successfully'
    if message is not None:
        stdout_msg = click.style(message, fg='yellow')

    click.echo(stdout_msg)
    click.echo('Try ' + click.style('stakkr services', fg='green') + ' to see new available services')


@stakkr.command(help="Update all services packs in services/")
@click.pass_context
def services_update(ctx):
    """See command Help."""
    from stakkr.services import update_all

    project_dir = _get_project_dir(ctx.obj['CONFIG'])
    services_dir = '{}/services'.format(project_dir)
    update_all(services_dir)
    print(click.style('Packages updated', fg='green'))


@stakkr.command(help="Start all (or a single as CONTAINER) container(s) defined in compose.ini")
@click.argument('container', required=False)
@click.option('--pull', '-p', help="Force a pull of the latest images versions", is_flag=True)
@click.option('--recreate', '-r', help="Recreate all containers", is_flag=True)
@click.option('--proxy/--no-proxy', '-P', help="Start proxy", default=True)
@click.pass_context
def start(ctx: Context, container: str, pull: bool, recreate: bool, proxy: bool):
    """See command Help."""
    print(click.style('[STARTING]', fg='green') + ' your stakkr services')

    ctx.obj['STAKKR'].start(container, pull, recreate, proxy)
    _show_status(ctx)


@stakkr.command(help="Display a list of running containers")
@click.pass_context
def status(ctx):
    """See command Help."""
    ctx.obj['STAKKR'].status()


@stakkr.command(help="Stop all (or a single as CONTAINER) container(s)")
@click.argument('container', required=False)
@click.option('--proxy/--no-proxy', '-P', help="Stop the proxy", default=True)
@click.pass_context
def stop(ctx: Context, container: str, proxy: bool):
    """See command Help."""
    print(click.style('[STOPPING]', fg='yellow') + ' your stakkr services')
    ctx.obj['STAKKR'].stop(container, proxy)


def _get_cmd_user(user: str, container: str):
    users = {'apache': 'www-data', 'nginx': 'www-data', 'php': 'www-data'}

    cmd_user = 'root' if user is None else user
    if container in users and user is None:
        cmd_user = users[container]

    return cmd_user


def _show_status(ctx):
    services_ports = ctx.obj['STAKKR'].get_services_urls()
    if services_ports == '':
        print('\nServices Status:')
        ctx.invoke(status)
        return

    print('\nServices URLs :')
    print(services_ports)


def _get_project_dir(config: str):
    from os.path import abspath, dirname

    if config is not None:
        config = abspath(config)
        return dirname(config)

    from stakkr.file_utils import find_project_dir
    return find_project_dir()


def debug_mode():
    """Guess if we are in debug mode, useful to display runtime errors."""
    if '--debug' in sys.argv or '-d' in sys.argv:
        return True

    return False


def run_commands(ctx: Context, extra_args: tuple, tty: bool):
    """Run commands for a specific alias"""
    commands = ctx.obj['STAKKR'].get_config()['aliases'][ctx.command.name]['exec']
    for command in commands:
        user = command['user'] if 'user' in command else 'root'
        container = command['container']
        args = command['args'] + list(extra_args) if extra_args is not None else []

        ctx.invoke(exec_cmd, user=user, container=container, command=args, tty=tty)


def main():
    """Call the CLI Script."""
    try:
        for alias, conf in get_aliases().items():
            if conf is None:
                continue

            cmd_help = conf['description'] if 'description' in conf else 'No description'

            @stakkr.command(help=cmd_help, name=alias)
            @click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
            @click.argument('extra_args', required=False, nargs=-1, type=click.UNPROCESSED)
            @click.pass_context
            def _f(ctx: Context, extra_args: tuple, tty: bool):
                """See command Help."""
                run_commands(ctx, extra_args, tty)

        stakkr(obj={})
    except Exception as error:
        msg = click.style(r""" ______ _____  _____   ____  _____
|  ____|  __ \|  __ \ / __ \|  __ \
| |__  | |__) | |__) | |  | | |__) |
|  __| |  _  /|  _  /| |  | |  _  /
| |____| | \ \| | \ \| |__| | | \ \
|______|_|  \_\_|  \_\\____/|_|  \_\

""", fg='yellow')
        msg += click.style('{}'.format(error), fg='red')
        print(msg + '\n', file=sys.stderr)

        if debug_mode() is True:
            raise error

        sys.exit(1)


if __name__ == '__main__':
    main()


1			#!/usr/bin/env python
2			# coding: utf-8
3			"""
4			CLI Entry Point.
5
6			From click, build stakkr.
7			Give all options to manage services to be launched, stopped, etc.
8			"""
9
10			import sys
11			import click
12			from click.core import Context
13			from stakkr.docker_actions import get_running_containers_names
14			from stakkr.aliases import get_aliases
15
16
17			@click.group(help="""Main CLI Tool that easily create / maintain
18			a stack of services, for example for web development.
19
20			Read the configuration file and setup the required services by
21			linking and managing everything for you.""")
22			@click.version_option('4.0.3')
23			@click.option('--config', '-c', help='Set the configuration filename (stakkr.yml by default)')
24			@click.option('--debug/--no-debug', '-d', default=False)
25			@click.option('--verbose', '-v', is_flag=True)
26			@click.pass_context
27			def stakkr(ctx: Context, config=None, debug=False, verbose=True):
28			"""Click group, set context and main object."""
29			from stakkr.actions import StakkrActions
30
31			ctx.obj['CONFIG'] = config
32			ctx.obj['DEBUG'] = debug
33			ctx.obj['VERBOSE'] = verbose
34			ctx.obj['STAKKR'] = StakkrActions(ctx.obj)
35
36
37			@stakkr.command(help="""Enter a container to perform direct actions such as
38			install packages, run commands, etc.""")
39			@click.argument('container', required=True)
40			@click.option('--user', '-u', help="User's name. Valid choices : www-data or root")
41			@click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
42			@click.pass_context
43			def console(ctx: Context, container: str, user: str, tty: bool):
44			"""See command Help."""
45			ctx.obj['STAKKR'].init_project()
46			ctx.obj['CTS'] = get_running_containers_names(ctx.obj['STAKKR'].project_name)
47			if ctx.obj['CTS']:
48			ct_choice = click.Choice(ctx.obj['CTS'])
49			ct_choice.convert(container, None, ctx)
50
51			ctx.obj['STAKKR'].console(container, _get_cmd_user(user, container), tty)
52
53
54			@stakkr.command(help="""Execute a command into a container.
55
56			Examples:\n
57			- ``stakkr -v exec mysql mysqldump -p'$MYSQL_ROOT_PASSWORD' mydb > /tmp/backup.sql``\n
58			- ``stakkr exec php php -v`` : Execute the php binary in the php container with option -v\n
59			- ``stakkr exec apache service apache2 restart``\n
60			""", name='exec', context_settings=dict(ignore_unknown_options=True, allow_interspersed_args=False))
61			@click.pass_context
62			@click.option('--user', '-u', help="User's name. Be careful, each container have its own users.")
63			@click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
64			@click.argument('container', required=True)
65			@click.argument('command', required=True, nargs=-1, type=click.UNPROCESSED)
66			def exec_cmd(ctx: Context, user: str, container: str, command: tuple, tty: bool):
67			"""See command Help."""
68			ctx.obj['STAKKR'].init_project()
69			ctx.obj['CTS'] = get_running_containers_names(ctx.obj['STAKKR'].project_name)
70			if ctx.obj['CTS']:
71			click.Choice(ctx.obj['CTS']).convert(container, None, ctx)
72
73			ctx.obj['STAKKR'].exec_cmd(container, _get_cmd_user(user, container), command, tty)
74
75
76			@stakkr.command(help="Restart all (or a single as CONTAINER) container(s)")
77			@click.argument('container', required=False)
78			@click.option('--pull', '-p', help="Force a pull of the latest images versions", is_flag=True)
79			@click.option('--recreate', '-r', help="Recreate all containers", is_flag=True)
80			@click.option('--proxy/--no-proxy', '-P', help="Restart the proxy", default=True)
81			@click.pass_context
82			def restart(ctx: Context, container: str, pull: bool, recreate: bool, proxy: bool):
83			"""See command Help."""
84			print(click.style('[RESTARTING]', fg='green') + ' your stakkr services')
85			try:
86			ctx.invoke(stop, container=container, proxy=proxy)
87			except Exception:
88			pass
89
90			ctx.invoke(start, container=container, pull=pull, recreate=recreate, proxy=proxy)
91
92
93			@stakkr.command(help="""List available services available for stakkr.yml
94			(with info if the service is enabled)""")
95			@click.pass_context
96			def services(ctx):
97			"""See command Help."""
98			ctx.obj['STAKKR'].init_project()
99
100			from stakkr.stakkr_compose import get_available_services
101
102			print('Available services usable in stakkr.yml ', end='')
103			print('({} = disabled) : '.format(click.style('✘', fg='red')))
104
105			svcs = ctx.obj['STAKKR'].get_config()['services']
106			enabled_svcs = [svc for svc, opts in svcs.items() if opts['enabled'] is True]
107			available_svcs = get_available_services(ctx.obj['STAKKR'].project_dir)
108			for available_svc in sorted(list(available_svcs.keys())):
109			sign = click.style('✘', fg='red')
110			if available_svc in enabled_svcs:
111			version = svcs[available_svc]['version']
112			sign = click.style(str(version), fg='green')
113
114			print(' - {} ({})'.format(available_svc, sign))
115
116
117			@stakkr.command(help="""Download a pack of services from github (see github) containing services.
118			PACKAGE is the git url or package name.
119			NAME is the directory name to clone the repo.
120			""")
121			@click.argument('package', required=True)
122			@click.argument('name', required=False)
123			@click.pass_context
124			def services_add(ctx: Context, package: str, name: str):
125			"""See command Help."""
126			from stakkr.services import install
127
128			project_dir = _get_project_dir(ctx.obj['CONFIG'])
129			services_dir = '{}/services'.format(project_dir)
130			name = package if name is None else name
131			success, message = install(services_dir, package, name)
132			if success is False:
133			click.echo(click.style(message, fg='red'))
134			sys.exit(1)
135
136			stdout_msg = 'Package "' + click.style(package, fg='green') + '" installed successfully'
137			if message is not None:
138			stdout_msg = click.style(message, fg='yellow')
139
140			click.echo(stdout_msg)
141			click.echo('Try ' + click.style('stakkr services', fg='green') + ' to see new available services')
142
143
144			@stakkr.command(help="Update all services packs in services/")
145			@click.pass_context
146			def services_update(ctx):
147			"""See command Help."""
148			from stakkr.services import update_all
149
150			project_dir = _get_project_dir(ctx.obj['CONFIG'])
151			services_dir = '{}/services'.format(project_dir)
152			update_all(services_dir)
153			print(click.style('Packages updated', fg='green'))
154
155
156			@stakkr.command(help="Start all (or a single as CONTAINER) container(s) defined in compose.ini")
157			@click.argument('container', required=False)
158			@click.option('--pull', '-p', help="Force a pull of the latest images versions", is_flag=True)
159			@click.option('--recreate', '-r', help="Recreate all containers", is_flag=True)
160			@click.option('--proxy/--no-proxy', '-P', help="Start proxy", default=True)
161			@click.pass_context
162			def start(ctx: Context, container: str, pull: bool, recreate: bool, proxy: bool):
163			"""See command Help."""
164			print(click.style('[STARTING]', fg='green') + ' your stakkr services')
165
166			ctx.obj['STAKKR'].start(container, pull, recreate, proxy)
167			_show_status(ctx)
168
169
170			@stakkr.command(help="Display a list of running containers")
171			@click.pass_context
172			def status(ctx):
173			"""See command Help."""
174			ctx.obj['STAKKR'].status()
175
176
177			@stakkr.command(help="Stop all (or a single as CONTAINER) container(s)")
178			@click.argument('container', required=False)
179			@click.option('--proxy/--no-proxy', '-P', help="Stop the proxy", default=True)
180			@click.pass_context
181			def stop(ctx: Context, container: str, proxy: bool):
182			"""See command Help."""
183			print(click.style('[STOPPING]', fg='yellow') + ' your stakkr services')
184			ctx.obj['STAKKR'].stop(container, proxy)
185
186
187			def _get_cmd_user(user: str, container: str):
188			users = {'apache': 'www-data', 'nginx': 'www-data', 'php': 'www-data'}
189
190			cmd_user = 'root' if user is None else user
191			if container in users and user is None:
192			cmd_user = users[container]
193
194			return cmd_user
195
196
197			def _show_status(ctx):
198			services_ports = ctx.obj['STAKKR'].get_services_urls()
199			if services_ports == '':
200			print('\nServices Status:')
201			ctx.invoke(status)
202			return
203
204			print('\nServices URLs :')
205			print(services_ports)
206
207
208			def _get_project_dir(config: str):
209			from os.path import abspath, dirname
210
211			if config is not None:
212			config = abspath(config)
213			return dirname(config)
214
215			from stakkr.file_utils import find_project_dir
216			return find_project_dir()
217
218
219			def debug_mode():
220			"""Guess if we are in debug mode, useful to display runtime errors."""
221			if '--debug' in sys.argv or '-d' in sys.argv:
222			return True
223
224			return False
225
226
227			def run_commands(ctx: Context, extra_args: tuple, tty: bool):
228			"""Run commands for a specific alias"""
229			commands = ctx.obj['STAKKR'].get_config()['aliases'][ctx.command.name]['exec']
230			for command in commands:
231			user = command['user'] if 'user' in command else 'root'
232			container = command['container']
233			args = command['args'] + list(extra_args) if extra_args is not None else []
234
235			ctx.invoke(exec_cmd, user=user, container=container, command=args, tty=tty)
236
237
238			def main():
239			"""Call the CLI Script."""
240			try:
241			for alias, conf in get_aliases().items():
242			if conf is None:
243			continue
244
245			cmd_help = conf['description'] if 'description' in conf else 'No description'
246
247			@stakkr.command(help=cmd_help, name=alias)
248			@click.option('--tty/--no-tty', '-t/ ', is_flag=True, default=True, help="Use a TTY")
249			@click.argument('extra_args', required=False, nargs=-1, type=click.UNPROCESSED)
250			@click.pass_context
251			def _f(ctx: Context, extra_args: tuple, tty: bool):
252			"""See command Help."""
253			run_commands(ctx, extra_args, tty)
254
255			stakkr(obj={})
256			except Exception as error:
257			msg = click.style(r""" ______ _____ _____ ____ _____
258			\| ____\| __ \\| __ \ / __ \\| __ \
259			\| \|__ \| \|__) \| \|__) \| \| \| \| \|__) \|
260			\| __\| \| _ /\| _ /\| \| \| \| _ /
261			\| \|____\| \| \ \\| \| \ \\| \|__\| \| \| \ \
262			\|______\|_\| \_\_\| \_\\____/\|_\| \_\
263
264			""", fg='yellow')
265			msg += click.style('{}'.format(error), fg='red')
266			print(msg + '\n', file=sys.stderr)
267
268			if debug_mode() is True:
269			raise error
270
271			sys.exit(1)
272
273
274			if __name__ == '__main__':
275			main()
276

stakkr-org / stakkr

Push — master ( 61e6a6...28b40d )

stakkr.cli.exec_cmd() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like