st2client.Client.__init__() - Code Metrics - Inspection of "Merge pull request #2450 from StackStorm/DZ/must_b..." - Plexxi/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 15192c...dfd3a9 )

by Plexxi

created 2016-02-03 06:42 UTC

st2client.Client.init() F

↳ Parent: st2client.Client

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
dl	0
loc	79
rs	3.8175
cc	9

How to fix Long Method

# Licensed to the StackStorm, Inc ('StackStorm') under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import os
import logging

import six

from st2client import models
from st2client.models.core import ResourceManager
from st2client.models.core import ActionAliasResourceManager
from st2client.models.core import LiveActionResourceManager
from st2client.models.core import TriggerInstanceResourceManager


LOG = logging.getLogger(__name__)

# Default values for the options not explicitly specified by the user
DEFAULT_API_PORT = 9101
DEFAULT_AUTH_PORT = 9100

DEFAULT_BASE_URL = 'http://127.0.0.1'
DEFAULT_API_VERSION = 'v1'


class Client(object):
    def __init__(self, base_url=None, auth_url=None, api_url=None, api_version=None, cacert=None,
                 debug=False, token=None):
        # Get CLI options. If not given, then try to get it from the environment.
        self.endpoints = dict()

        # Populate the endpoints
        if base_url:
            self.endpoints['base'] = base_url
        else:
            self.endpoints['base'] = os.environ.get('ST2_BASE_URL', DEFAULT_BASE_URL)

        api_version = api_version or os.environ.get('ST2_API_VERSION', DEFAULT_API_VERSION)

        if api_url:
            self.endpoints['api'] = api_url
        else:
            self.endpoints['api'] = os.environ.get(
                'ST2_API_URL', '%s:%s/%s' % (self.endpoints['base'], DEFAULT_API_PORT, api_version))

        if auth_url:
            self.endpoints['auth'] = auth_url
        else:
            self.endpoints['auth'] = os.environ.get(
                'ST2_AUTH_URL', '%s:%s' % (self.endpoints['base'], DEFAULT_AUTH_PORT))

        if cacert is not None:
            self.cacert = cacert
        else:
            self.cacert = os.environ.get('ST2_CACERT', None)

        # Note: boolean is also a valid value for "cacert"
        is_cacert_string = isinstance(self.cacert, six.string_types)
        if (self.cacert and is_cacert_string and not os.path.isfile(self.cacert)):

            raise ValueError('CA cert file "%s" does not exist.' % (self.cacert))

        self.debug = debug

        # Note: This is a nasty hack for now, but we need to get rid of the decrator abuse
        if token:
            os.environ['ST2_AUTH_TOKEN'] = token

        self.token = token

        # Instantiate resource managers and assign appropriate API endpoint.
        self.managers = dict()
        self.managers['Token'] = ResourceManager(
            models.Token, self.endpoints['auth'], cacert=self.cacert, debug=self.debug)
        self.managers['RunnerType'] = ResourceManager(
            models.RunnerType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Action'] = ResourceManager(
            models.Action, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['ActionAlias'] = ActionAliasResourceManager(
            models.ActionAlias, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['ApiKey'] = ResourceManager(
            models.ApiKey, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['LiveAction'] = LiveActionResourceManager(
            models.LiveAction, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Policy'] = ResourceManager(
            models.Policy, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['PolicyType'] = ResourceManager(
            models.PolicyType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Rule'] = ResourceManager(
            models.Rule, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Sensor'] = ResourceManager(
            models.Sensor, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['TriggerType'] = ResourceManager(
            models.TriggerType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Trigger'] = ResourceManager(
            models.Trigger, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['TriggerInstance'] = TriggerInstanceResourceManager(
            models.TriggerInstance, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['KeyValuePair'] = ResourceManager(
            models.KeyValuePair, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Webhook'] = ResourceManager(
            models.Webhook, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['Trace'] = ResourceManager(
            models.Trace, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
        self.managers['RuleEnforcement'] = ResourceManager(
            models.RuleEnforcement, self.endpoints['api'], cacert=self.cacert, debug=self.debug)

    @property
    def actions(self):
        return self.managers['Action']

    @property
    def apikeys(self):
        return self.managers['ApiKey']

    @property
    def keys(self):
        return self.managers['KeyValuePair']

    @property
    def liveactions(self):
        return self.managers['LiveAction']

    @property
    def policies(self):
        return self.managers['Policy']

    @property
    def policytypes(self):
        return self.managers['PolicyType']

    @property
    def rules(self):
        return self.managers['Rule']

    @property
    def runners(self):
        return self.managers['RunnerType']

    @property
    def sensors(self):
        return self.managers['Sensor']

    @property
    def tokens(self):
        return self.managers['Token']

    @property
    def triggertypes(self):
        return self.managers['TriggerType']

    @property
    def triggerinstances(self):
        return self.managers['TriggerInstance']

    @property
    def trace(self):
        return self.managers['Trace']

    @property
    def ruleenforcements(self):
        return self.managers['RuleEnforcement']


1			# Licensed to the StackStorm, Inc ('StackStorm') under one or more
2			# contributor license agreements. See the NOTICE file distributed with
3			# this work for additional information regarding copyright ownership.
4			# The ASF licenses this file to You under the Apache License, Version 2.0
5			# (the "License"); you may not use this file except in compliance with
6			# the License. You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15
16			import os
17			import logging
18
19			import six
20
21			from st2client import models
22			from st2client.models.core import ResourceManager
23			from st2client.models.core import ActionAliasResourceManager
24			from st2client.models.core import LiveActionResourceManager
25			from st2client.models.core import TriggerInstanceResourceManager
26
27
28			LOG = logging.getLogger(__name__)
29
30			# Default values for the options not explicitly specified by the user
31			DEFAULT_API_PORT = 9101
32			DEFAULT_AUTH_PORT = 9100
33
34			DEFAULT_BASE_URL = 'http://127.0.0.1'
35			DEFAULT_API_VERSION = 'v1'
36
37
38			class Client(object):
39			def __init__(self, base_url=None, auth_url=None, api_url=None, api_version=None, cacert=None,
40			debug=False, token=None):
41			# Get CLI options. If not given, then try to get it from the environment.
42			self.endpoints = dict()
43
44			# Populate the endpoints
45			if base_url:
46			self.endpoints['base'] = base_url
47			else:
48			self.endpoints['base'] = os.environ.get('ST2_BASE_URL', DEFAULT_BASE_URL)
49
50			api_version = api_version or os.environ.get('ST2_API_VERSION', DEFAULT_API_VERSION)
51
52			if api_url:
53			self.endpoints['api'] = api_url
54			else:
55			self.endpoints['api'] = os.environ.get(
56			'ST2_API_URL', '%s:%s/%s' % (self.endpoints['base'], DEFAULT_API_PORT, api_version))
57
58			if auth_url:
59			self.endpoints['auth'] = auth_url
60			else:
61			self.endpoints['auth'] = os.environ.get(
62			'ST2_AUTH_URL', '%s:%s' % (self.endpoints['base'], DEFAULT_AUTH_PORT))
63
64			if cacert is not None:
65			self.cacert = cacert
66			else:
67			self.cacert = os.environ.get('ST2_CACERT', None)
68
69			# Note: boolean is also a valid value for "cacert"
70			is_cacert_string = isinstance(self.cacert, six.string_types)
71			if (self.cacert and is_cacert_string and not os.path.isfile(self.cacert)):
			0 ignored issues – show Unused Code Coding Style introduced 2016-02-02 17:21 UTC by Report Bug Copy Issue Report There is an unnecessary parenthesis after `if`. Loading history...
72			raise ValueError('CA cert file "%s" does not exist.' % (self.cacert))
73
74			self.debug = debug
75
76			# Note: This is a nasty hack for now, but we need to get rid of the decrator abuse
77			if token:
78			os.environ['ST2_AUTH_TOKEN'] = token
79
80			self.token = token
81
82			# Instantiate resource managers and assign appropriate API endpoint.
83			self.managers = dict()
84			self.managers['Token'] = ResourceManager(
85			models.Token, self.endpoints['auth'], cacert=self.cacert, debug=self.debug)
86			self.managers['RunnerType'] = ResourceManager(
87			models.RunnerType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
88			self.managers['Action'] = ResourceManager(
89			models.Action, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
90			self.managers['ActionAlias'] = ActionAliasResourceManager(
91			models.ActionAlias, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
92			self.managers['ApiKey'] = ResourceManager(
93			models.ApiKey, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
94			self.managers['LiveAction'] = LiveActionResourceManager(
95			models.LiveAction, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
96			self.managers['Policy'] = ResourceManager(
97			models.Policy, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
98			self.managers['PolicyType'] = ResourceManager(
99			models.PolicyType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
100			self.managers['Rule'] = ResourceManager(
101			models.Rule, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
102			self.managers['Sensor'] = ResourceManager(
103			models.Sensor, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
104			self.managers['TriggerType'] = ResourceManager(
105			models.TriggerType, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
106			self.managers['Trigger'] = ResourceManager(
107			models.Trigger, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
108			self.managers['TriggerInstance'] = TriggerInstanceResourceManager(
109			models.TriggerInstance, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
110			self.managers['KeyValuePair'] = ResourceManager(
111			models.KeyValuePair, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
112			self.managers['Webhook'] = ResourceManager(
113			models.Webhook, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
114			self.managers['Trace'] = ResourceManager(
115			models.Trace, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
116			self.managers['RuleEnforcement'] = ResourceManager(
117			models.RuleEnforcement, self.endpoints['api'], cacert=self.cacert, debug=self.debug)
118
119			@property
120			def actions(self):
121			return self.managers['Action']
122
123			@property
124			def apikeys(self):
125			return self.managers['ApiKey']
126
127			@property
128			def keys(self):
129			return self.managers['KeyValuePair']
130
131			@property
132			def liveactions(self):
133			return self.managers['LiveAction']
134
135			@property
136			def policies(self):
137			return self.managers['Policy']
138
139			@property
140			def policytypes(self):
141			return self.managers['PolicyType']
142
143			@property
144			def rules(self):
145			return self.managers['Rule']
146
147			@property
148			def runners(self):
149			return self.managers['RunnerType']
150
151			@property
152			def sensors(self):
153			return self.managers['Sensor']
154
155			@property
156			def tokens(self):
157			return self.managers['Token']
158
159			@property
160			def triggertypes(self):
161			return self.managers['TriggerType']
162
163			@property
164			def triggerinstances(self):
165			return self.managers['TriggerInstance']
166
167			@property
168			def trace(self):
169			return self.managers['Trace']
170
171			@property
172			def ruleenforcements(self):
173			return self.managers['RuleEnforcement']
174

Plexxi / st2

GitHub Access Token became invalid

Push — master ( 15192c...dfd3a9 )

st2client.Client.__init__() F

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like

st2client.Client.init() F