publish_sqs() - Code Metrics - Inspection of "Add sns message translator (#9)" - georgeyk/loafer - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d20e04...b19ded )

by George

created 2016-04-20 12:27 UTC

publish_sqs() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	7
rs	9.4285

# -*- coding: utf-8 -*-
# vi:si:et:sw=4:sts=4:ts=4

import json
import logging

import click
# .scrutinizer.yml
before_commands:
    - sudo pip install abc # Python2
    - sudo pip3 install abc # Python3

from . import __version__
from .conf import settings
from .manager import LoaferManager
from .aws.publisher import sqs_publish, sns_publish

logger = logging.getLogger(__name__)



def _bootstrap():
    logging.basicConfig(level=settings.LOAFER_LOGLEVEL,
                        format=settings.LOAFER_LOG_FORMAT)

    click.echo('>. Version: {}'.format(__version__))
    click.echo('>. Maximum concurrent jobs: {}'.format(settings.LOAFER_MAX_JOBS))
    click.echo('>. Routes:')
    for route in settings.LOAFER_ROUTES:
        click.echo('>.\t{}:'.format(route['name']))
        click.echo('>.\t\tSource: {}'.format(route['source']))
        click.echo('>.\t\tHandler: {}'.format(route['handler']))
        click.echo('>.\t\tMessage Translator: {}'.format(route['message_translator']))
        click.echo('>.\t\tConsumer: {}'.format(settings.LOAFER_DEFAULT_CONSUMER_CLASS))
        click.echo('>.\t\tConsumer Options: {}'.format(settings.LOAFER_DEFAULT_CONSUMER_OPTIONS))


def main(**kwargs):

    click.secho('>. Starting Loafer ...', bold=True, fg='green')

    _bootstrap()

    click.secho('>. Hit CTRL-C to stop', bold=True, fg='yellow')

    loafer = LoaferManager()
    loafer.start()


#
# CLI
#

CLICK_CONTEXT_SETTINGS = {'help_option_names': ['-h', '--help']}


def show_version(ctx, param, value):

    """Show Loafer version"""
    if not value or ctx.resilient_parsing:
        return

    click.echo(__version__)
    ctx.exit()


@click.group(invoke_without_command=True,
             context_settings=CLICK_CONTEXT_SETTINGS)
@click.option('-v', default=False, is_flag=True,
              help='Verbose mode (set LOAFER_LOGLEVEL=INFO)')
@click.option('-vv', default=False, is_flag=True,
              help='Very verbose mode (set LOAFER_LOGLEVEL=DEBUG)')
@click.option('--version', is_flag=True, is_eager=True, expose_value=False,
              callback=show_version, help="Show Loafer's version and exit")
@click.option('--max-jobs', default=None, type=int,
              help='Maximum concurrent jobs, overrides LOAFER_MAX_JOBS')
@click.option('--max-threads', default=None, type=int,
              help='Maximum threads, overrides LOAFER_MAX_THREAD_POOL')
@click.option('--source', default=None,
              help='The route source, updates the default route source')
@click.option('--handler', default=None,
              help='The route handler, updates the default route handler')
@click.option('--translator', default=None,
              help='The message translator class, updates the default route message translator')
@click.option('--consumer', default=None,
              help='The consumer class, overrides LOAFER_DEFAULT_CONSUMER_CLASS')
@click.option('--consumer-opts', default=None,
              help='The consumer options (assumes json), overrides LOAFER_DEFAULT_CONSUMER_OPTIONS')
@click.pass_context
def cli(context, v, vv, max_jobs, max_threads, source, handler, translator,

        consumer, consumer_opts):
    if v:
        settings.LOAFER_LOGLEVEL = 'INFO'
    if vv:
        settings.LOAFER_LOGLEVEL = 'DEBUG'
    if max_jobs and max_jobs >= 1:
        settings.LOAFER_MAX_JOBS = max_jobs
    if max_threads and max_threads >= 1:
        settings.LOAFER_MAX_THREAD_POOL = max_threads
    if source:
        settings.LOAFER_ROUTES[0]['source'] = source
    if handler:
        settings.LOAFER_ROUTES[0]['handler'] = handler
    if translator:
        settings.LOAFER_ROUTES[0]['message_translator'] = translator
    if consumer:
        settings.LOAFER_DEFAULT_CONSUMER_CLASS = consumer
    if consumer_opts:
        opts = json.loads(consumer_opts)
        settings.LOAFER_DEFAULT_CONSUMER_OPTIONS = opts

    if context.invoked_subcommand is None:
        main()


@cli.command()
@click.option('--queue', default=None, help='SQS queue name ou url')
@click.option('--msg', help='Message to publish (assumes json format)')
def publish_sqs(queue, msg):
    """Publish messages to AWS SQS"""
    response = sqs_publish(queue, json.dumps(msg))
    click.echo('Response: {}'.format(response))


@cli.command()
@click.option('--topic', default=None, help='SNS topic name ou arn')
@click.option('--msg', help='Message to publish (assumes json format)')
def publish_sns(topic, msg):
    """Publish messages to AWS SNS"""
    # We need this validation because our sns_publish always use json format
    try:
        json.loads(msg)
    except json.decoder.JSONDecodeError:

        click.secho('"{}" should be a valid json'.format(msg), fg='red')
    else:
        response = sns_publish(topic, msg)
        click.echo('Response: {}'.format(response))


1			# -- coding: utf-8 --
2			# vi:si:et:sw=4:sts=4:ts=4
3
4			import json
5			import logging
6
7			import click
			1 ignored issue – show Configuration introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The import `click` could not be resolved. This can be caused by one of the following: 1. Missing Dependencies This error could indicate a configuration issue of Pylint. Make sure that your libraries are available by adding the necessary commands. # .scrutinizer.yml before_commands: - sudo pip install abc # Python2 - sudo pip3 install abc # Python3 Tip: We are currently not using virtualenv to run pylint, when installing your modules make sure to use the command for the correct version. 2. Missing __init__.py files This error could also result from missing `__init__.py` files in your module folders. Make sure that you place one file in each sub-folder. Loading history...
8
9			from . import __version__
10			from .conf import settings
11			from .manager import LoaferManager
12			from .aws.publisher import sqs_publish, sns_publish
13
14			logger = logging.getLogger(__name__)
			1 ignored issue – show Coding Style Naming introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The name `logger` does not conform to the constant naming conventions (`(([A-Z_][A-Z0-9_])\|(__.__))$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
15
16
17			def _bootstrap():
18			logging.basicConfig(level=settings.LOAFER_LOGLEVEL,
19			format=settings.LOAFER_LOG_FORMAT)
20
21			click.echo('>. Version: {}'.format(__version__))
22			click.echo('>. Maximum concurrent jobs: {}'.format(settings.LOAFER_MAX_JOBS))
23			click.echo('>. Routes:')
24			for route in settings.LOAFER_ROUTES:
25			click.echo('>.\t{}:'.format(route['name']))
26			click.echo('>.\t\tSource: {}'.format(route['source']))
27			click.echo('>.\t\tHandler: {}'.format(route['handler']))
28			click.echo('>.\t\tMessage Translator: {}'.format(route['message_translator']))
29			click.echo('>.\t\tConsumer: {}'.format(settings.LOAFER_DEFAULT_CONSUMER_CLASS))
30			click.echo('>.\t\tConsumer Options: {}'.format(settings.LOAFER_DEFAULT_CONSUMER_OPTIONS))
31
32
33			def main(**kwargs):
			0 ignored issues – show Unused Code introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The argument `kwargs` seems to be unused. Loading history...
34			click.secho('>. Starting Loafer ...', bold=True, fg='green')
35
36			_bootstrap()
37
38			click.secho('>. Hit CTRL-C to stop', bold=True, fg='yellow')
39
40			loafer = LoaferManager()
41			loafer.start()
42
43
44			#
45			# CLI
46			#
47
48			CLICK_CONTEXT_SETTINGS = {'help_option_names': ['-h', '--help']}
49
50
51			def show_version(ctx, param, value):
			0 ignored issues – show Unused Code introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The argument `param` seems to be unused. Loading history...
52			"""Show Loafer version"""
53			if not value or ctx.resilient_parsing:
54			return
55
56			click.echo(__version__)
57			ctx.exit()
58
59
60			@click.group(invoke_without_command=True,
61			context_settings=CLICK_CONTEXT_SETTINGS)
62			@click.option('-v', default=False, is_flag=True,
63			help='Verbose mode (set LOAFER_LOGLEVEL=INFO)')
64			@click.option('-vv', default=False, is_flag=True,
65			help='Very verbose mode (set LOAFER_LOGLEVEL=DEBUG)')
66			@click.option('--version', is_flag=True, is_eager=True, expose_value=False,
67			callback=show_version, help="Show Loafer's version and exit")
68			@click.option('--max-jobs', default=None, type=int,
69			help='Maximum concurrent jobs, overrides LOAFER_MAX_JOBS')
70			@click.option('--max-threads', default=None, type=int,
71			help='Maximum threads, overrides LOAFER_MAX_THREAD_POOL')
72			@click.option('--source', default=None,
73			help='The route source, updates the default route source')
74			@click.option('--handler', default=None,
75			help='The route handler, updates the default route handler')
76			@click.option('--translator', default=None,
77			help='The message translator class, updates the default route message translator')
78			@click.option('--consumer', default=None,
79			help='The consumer class, overrides LOAFER_DEFAULT_CONSUMER_CLASS')
80			@click.option('--consumer-opts', default=None,
81			help='The consumer options (assumes json), overrides LOAFER_DEFAULT_CONSUMER_OPTIONS')
82			@click.pass_context
83			def cli(context, v, vv, max_jobs, max_threads, source, handler, translator,
			2 ignored issues – show Coding Style Naming introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The name `v` does not conform to the argument naming conventions (`[a-z_][a-z0-9_]{2,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history... Coding Style Naming introduced 2016-04-20 00:12 UTC by Report Bug Copy Issue Report The name `vv` does not conform to the argument naming conventions (`[a-z_][a-z0-9_]{2,30}$`). This check looks for invalid names for a range of different identifiers. You can set regular expressions to which the identifiers must conform if the defaults do not match your requirements. If your project includes a Pylint configuration file, the settings contained in that file take precedence. To find out more about Pylint, please refer to their site. Loading history...
84			consumer, consumer_opts):
85			if v:
86			settings.LOAFER_LOGLEVEL = 'INFO'
87			if vv:
88			settings.LOAFER_LOGLEVEL = 'DEBUG'
89			if max_jobs and max_jobs >= 1:
90			settings.LOAFER_MAX_JOBS = max_jobs
91			if max_threads and max_threads >= 1:
92			settings.LOAFER_MAX_THREAD_POOL = max_threads
93			if source:
94			settings.LOAFER_ROUTES[0]['source'] = source
95			if handler:
96			settings.LOAFER_ROUTES[0]['handler'] = handler
97			if translator:
98			settings.LOAFER_ROUTES[0]['message_translator'] = translator
99			if consumer:
100			settings.LOAFER_DEFAULT_CONSUMER_CLASS = consumer
101			if consumer_opts:
102			opts = json.loads(consumer_opts)
103			settings.LOAFER_DEFAULT_CONSUMER_OPTIONS = opts
104
105			if context.invoked_subcommand is None:
106			main()
107
108
109			@cli.command()
110			@click.option('--queue', default=None, help='SQS queue name ou url')
111			@click.option('--msg', help='Message to publish (assumes json format)')
112			def publish_sqs(queue, msg):
113			"""Publish messages to AWS SQS"""
114			response = sqs_publish(queue, json.dumps(msg))
115			click.echo('Response: {}'.format(response))
116
117
118			@cli.command()
119			@click.option('--topic', default=None, help='SNS topic name ou arn')
120			@click.option('--msg', help='Message to publish (assumes json format)')
121			def publish_sns(topic, msg):
122			"""Publish messages to AWS SNS"""
123			# We need this validation because our sns_publish always use json format
124			try:
125			json.loads(msg)
126			except json.decoder.JSONDecodeError:
			1 ignored issue – show Bug introduced 2016-04-20 12:24 UTC by Report Bug Copy Issue Report The Module `json.decoder` does not seem to have a member named `JSONDecodeError`. This check looks for calls to members that are non-existent. These calls will fail. The member could have been renamed or removed. Loading history...
127			click.secho('"{}" should be a valid json'.format(msg), fg='red')
128			else:
129			response = sns_publish(topic, msg)
130			click.echo('Response: {}'.format(response))
131

georgeyk / loafer

Push — master ( d20e04...b19ded )

publish_sqs() A

Complexity

Size

Duplication

1. Missing Dependencies

2. Missing __init__.py files

Duplication Side-by-Side

Filter issues like

2. Missing init.py files