bin.Worker - Code Metrics - Inspection of "Added WORKFLOW_COMPLETED message for SQS" - sportarchive/CloudProcessingEngine-Decider - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#35)

by Tyler

created 2016-05-20 20:56 UTC

bin.Worker A

↳ Parent: Project

Complexity

Total Complexity

Size/Duplication

Total Lines	15
Duplicated Lines	0 %

Metric	Value
dl	0
loc	15
rs	10
wmc	3

#!/usr/bin/env python

from __future__ import (
    absolute_import,
    division,
    print_function
)

import argparse
import logging
import os
import sys

import boto.swf.layer2 as swf

sys.path.append(
    os.path.realpath(
        os.path.join(
            os.path.dirname(__file__),
            '..'
        )
    )
)

from pydecider.register import register


class Worker(swf.ActivityWorker):

    version = 1.0

    def __init__(self, domain, task_list):
        self.domain = domain
        self.task_list = task_list
        super(Worker, self).__init__()

    def run(self):
        activity_task = self.poll()
        if 'activityId' in activity_task:
            print('%s: %r' % (self.task_list, activity_task))
            self.complete()
        return True


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--domain', required=True)
    parser.add_argument('--task_list', required=True)
    parser.add_argument('--activity', required=True)
    parser.add_argument('--version', required=True)
    args = parser.parse_args()

    register(activities=[(args.activity, args.version)])

    worker = Worker(args.domain, args.task_list)

    while worker.run():
        pass


if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)

    main()


1			#!/usr/bin/env python
2
3			from __future__ import (
4			absolute_import,
5			division,
6			print_function
7			)
8
9			import argparse
10			import logging
11			import os
12			import sys
13
14			import boto.swf.layer2 as swf
15
16			sys.path.append(
17			os.path.realpath(
18			os.path.join(
19			os.path.dirname(__file__),
20			'..'
21			)
22			)
23			)
24
25			from pydecider.register import register
26
27
28			class Worker(swf.ActivityWorker):
29
30			version = 1.0
31
32			def __init__(self, domain, task_list):
33			self.domain = domain
34			self.task_list = task_list
35			super(Worker, self).__init__()
36
37			def run(self):
38			activity_task = self.poll()
39			if 'activityId' in activity_task:
40			print('%s: %r' % (self.task_list, activity_task))
41			self.complete()
42			return True
43
44
45			def main():
46			parser = argparse.ArgumentParser()
47			parser.add_argument('--domain', required=True)
48			parser.add_argument('--task_list', required=True)
49			parser.add_argument('--activity', required=True)
50			parser.add_argument('--version', required=True)
51			args = parser.parse_args()
52
53			register(activities=[(args.activity, args.version)])
54
55			worker = Worker(args.domain, args.task_list)
56
57			while worker.run():
58			pass
59
60
61			if __name__ == '__main__':
62			logging.basicConfig(level=logging.DEBUG)
63
64			main()
65

sportarchive / CloudProcessingEngine-Decider

Pull Request — master (#35)

bin.Worker A

Complexity

Size/Duplication

Duplication Side-by-Side

Filter issues like