InquiryAPI.from_model() - Code Metrics - Inspection of "Merge pull request #3653 from StackStorm/api-ask-r..." - StackStorm/st2 - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( ac1734...fff02f )

unknown

created 2017-10-03 10:00 UTC

InquiryAPI.from_model() A

↳ Parent: InquiryAPI

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	2
dl	0
loc	17
rs	9.4285
c	2
b	0
f	0

# 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 copy
import six

from st2common.constants.action import LIVEACTION_STATUSES
from st2common.models.api.base import BaseAPI
from st2common.models.api.action import RunnerTypeAPI, ActionAPI, LiveActionAPI
from st2common.models.db.execution import ActionExecutionDB
from st2common import log as logging


LOG = logging.getLogger(__name__)

REQUIRED_ATTR_SCHEMAS = {
    "action": copy.deepcopy(ActionAPI.schema),
    "runner": copy.deepcopy(RunnerTypeAPI.schema),
    "liveaction": copy.deepcopy(LiveActionAPI.schema),
}

for k, v in six.iteritems(REQUIRED_ATTR_SCHEMAS):
    v.update({"required": True})


class InquiryAPI(BaseAPI):
    """Model for working with Inquiries within the API controller

    Please see InquiryResponseAPI for a model more appropriate for API
    responses. This model contains fields that do not comply with the API
    spec.
    """

    # Inquiries don't have their own database model, since (right now) they are just
    # ActionExecutions with a "pending" status. The "from_model" function will serve
    # to transform the relevant fields to a new InquiryAPI instance
    model = ActionExecutionDB

    schema = {
        "title": "Inquiry",
        "description": "Record of an Inquiry",
        "type": "object",
        "properties": {
            "id": {
                "type": "string",
                "required": True
            },
            "route": {
                "type": "string",
                "default": "",
                "required": True
            },
            "ttl": {
                "type": "integer",
                "default": 1440,
                "required": True
            },
            "users": {
                "type": "array",
                "default": [],
                "required": True
            },
            "roles": {
                "type": "array",
                "default": [],
                "required": True
            },
            "schema": {
                "type": "object",
                "default": {
                    "title": "response_data",
                    "type": "object",
                    "properties": {
                        "continue": {
                            "type": "boolean",
                            "description": "Would you like to continue the workflow?",
                            "required": True
                        }
                    },
                },
                "required": True
            },
            "liveaction": REQUIRED_ATTR_SCHEMAS['liveaction'],
            "runner": REQUIRED_ATTR_SCHEMAS['runner'],
            "status": {
                "description": "The current status of the action execution.",
                "type": "string",
                "enum": LIVEACTION_STATUSES
            },
            "parent": {"type": "string"},
            "result": {
                "anyOf": [{"type": "array"},
                          {"type": "boolean"},
                          {"type": "integer"},
                          {"type": "number"},
                          {"type": "object"},
                          {"type": "string"}]
            }
        },
        "additionalProperties": False
    }

    @classmethod
    def from_model(cls, model, mask_secrets=False):
        doc = cls._from_model(model, mask_secrets=mask_secrets)

        newdoc = {
            "id": doc["id"],
            "runner": doc["runner"],
            "status": doc["status"],
            "liveaction": doc["liveaction"],
            "parent": doc.get("parent"),
            "result": doc['result']
        }

        for field in ["route", "ttl", "users", "roles", "schema"]:
            newdoc[field] = doc["result"].get(field)

        return cls(**newdoc)


class InquiryResponseAPI(BaseAPI):
    """A more pruned Inquiry model, containing only the fields needed for an API response
    """

    model = ActionExecutionDB
    schema = {
        "title": "Inquiry",
        "description": "Record of an Inquiry",
        "type": "object",
        "properties": {
            "id": {
                "type": "string",
                "required": True
            },
            "route": {
                "type": "string",
                "default": "",
                "required": True
            },
            "ttl": {
                "type": "integer",
                "default": 1440,
                "required": True
            },
            "users": {
                "type": "array",
                "default": [],
                "required": True
            },
            "roles": {
                "type": "array",
                "default": [],
                "required": True
            },
            "schema": {
                "type": "object",
                "default": {
                    "title": "response_data",
                    "type": "object",
                    "properties": {
                        "continue": {
                            "type": "boolean",
                            "description": "Would you like to continue the workflow?",
                            "required": True
                        }
                    },
                },
                "required": True
            }
        },
        "additionalProperties": False
    }

    @classmethod
    def from_model(cls, model, mask_secrets=False, skip_db=False):

        """Create InquiryResponseAPI instance from model

        Allows skipping the BaseAPI._from_model function if you already
        have a properly formed dict and just need to prune it

        :param skip_db: Skip the parent class' _from_model function call
        :rtype: InquiryResponseAPI
        """

        if not skip_db:
            doc = cls._from_model(model, mask_secrets=mask_secrets)
        else:
            doc = model

        newdoc = {
            "id": doc["id"]
        }
        for field in ["route", "ttl", "users", "roles", "schema"]:
            newdoc[field] = doc["result"].get(field)

        return cls(**newdoc)

    @classmethod
    def from_inquiry_api(cls, inquiry_api, mask_secrets=False):
        """ Allows translation of InquiryAPI directly to InquiryResponseAPI

        This bypasses the DB modeling, since there's no DB model for Inquiries yet.
        """

        return cls(
            id=getattr(inquiry_api, 'id', None),
            route=getattr(inquiry_api, 'route', None),
            ttl=getattr(inquiry_api, 'ttl', None),
            users=getattr(inquiry_api, 'users', None),
            roles=getattr(inquiry_api, 'roles', None),
            schema=getattr(inquiry_api, 'schema', None)
        )


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 copy
17			import six
18
19			from st2common.constants.action import LIVEACTION_STATUSES
20			from st2common.models.api.base import BaseAPI
21			from st2common.models.api.action import RunnerTypeAPI, ActionAPI, LiveActionAPI
22			from st2common.models.db.execution import ActionExecutionDB
23			from st2common import log as logging
24
25
26			LOG = logging.getLogger(__name__)
27
28			REQUIRED_ATTR_SCHEMAS = {
29			"action": copy.deepcopy(ActionAPI.schema),
30			"runner": copy.deepcopy(RunnerTypeAPI.schema),
31			"liveaction": copy.deepcopy(LiveActionAPI.schema),
32			}
33
34			for k, v in six.iteritems(REQUIRED_ATTR_SCHEMAS):
35			v.update({"required": True})
36
37
38			class InquiryAPI(BaseAPI):
39			"""Model for working with Inquiries within the API controller
40
41			Please see InquiryResponseAPI for a model more appropriate for API
42			responses. This model contains fields that do not comply with the API
43			spec.
44			"""
45
46			# Inquiries don't have their own database model, since (right now) they are just
47			# ActionExecutions with a "pending" status. The "from_model" function will serve
48			# to transform the relevant fields to a new InquiryAPI instance
49			model = ActionExecutionDB
50
51			schema = {
52			"title": "Inquiry",
53			"description": "Record of an Inquiry",
54			"type": "object",
55			"properties": {
56			"id": {
57			"type": "string",
58			"required": True
59			},
60			"route": {
61			"type": "string",
62			"default": "",
63			"required": True
64			},
65			"ttl": {
66			"type": "integer",
67			"default": 1440,
68			"required": True
69			},
70			"users": {
71			"type": "array",
72			"default": [],
73			"required": True
74			},
75			"roles": {
76			"type": "array",
77			"default": [],
78			"required": True
79			},
80			"schema": {
81			"type": "object",
82			"default": {
83			"title": "response_data",
84			"type": "object",
85			"properties": {
86			"continue": {
87			"type": "boolean",
88			"description": "Would you like to continue the workflow?",
89			"required": True
90			}
91			},
92			},
93			"required": True
94			},
95			"liveaction": REQUIRED_ATTR_SCHEMAS['liveaction'],
96			"runner": REQUIRED_ATTR_SCHEMAS['runner'],
97			"status": {
98			"description": "The current status of the action execution.",
99			"type": "string",
100			"enum": LIVEACTION_STATUSES
101			},
102			"parent": {"type": "string"},
103			"result": {
104			"anyOf": [{"type": "array"},
105			{"type": "boolean"},
106			{"type": "integer"},
107			{"type": "number"},
108			{"type": "object"},
109			{"type": "string"}]
110			}
111			},
112			"additionalProperties": False
113			}
114
115			@classmethod
116			def from_model(cls, model, mask_secrets=False):
117			doc = cls._from_model(model, mask_secrets=mask_secrets)
118
119			newdoc = {
120			"id": doc["id"],
121			"runner": doc["runner"],
122			"status": doc["status"],
123			"liveaction": doc["liveaction"],
124			"parent": doc.get("parent"),
125			"result": doc['result']
126			}
127
128			for field in ["route", "ttl", "users", "roles", "schema"]:
129			newdoc[field] = doc["result"].get(field)
130
131			return cls(**newdoc)
132
133
134			class InquiryResponseAPI(BaseAPI):
135			"""A more pruned Inquiry model, containing only the fields needed for an API response
136			"""
137
138			model = ActionExecutionDB
139			schema = {
140			"title": "Inquiry",
141			"description": "Record of an Inquiry",
142			"type": "object",
143			"properties": {
144			"id": {
145			"type": "string",
146			"required": True
147			},
148			"route": {
149			"type": "string",
150			"default": "",
151			"required": True
152			},
153			"ttl": {
154			"type": "integer",
155			"default": 1440,
156			"required": True
157			},
158			"users": {
159			"type": "array",
160			"default": [],
161			"required": True
162			},
163			"roles": {
164			"type": "array",
165			"default": [],
166			"required": True
167			},
168			"schema": {
169			"type": "object",
170			"default": {
171			"title": "response_data",
172			"type": "object",
173			"properties": {
174			"continue": {
175			"type": "boolean",
176			"description": "Would you like to continue the workflow?",
177			"required": True
178			}
179			},
180			},
181			"required": True
182			}
183			},
184			"additionalProperties": False
185			}
186
187			@classmethod
188			def from_model(cls, model, mask_secrets=False, skip_db=False):
			0 ignored issues – show Bug introduced 2017-09-30 01:06 UTC by Report Bug Copy Issue Report Arguments number differs from overridden 'from_model' method Loading history...
189			"""Create InquiryResponseAPI instance from model
190
191			Allows skipping the BaseAPI._from_model function if you already
192			have a properly formed dict and just need to prune it
193
194			:param skip_db: Skip the parent class' _from_model function call
195			:rtype: InquiryResponseAPI
196			"""
197
198			if not skip_db:
199			doc = cls._from_model(model, mask_secrets=mask_secrets)
200			else:
201			doc = model
202
203			newdoc = {
204			"id": doc["id"]
205			}
206			for field in ["route", "ttl", "users", "roles", "schema"]:
207			newdoc[field] = doc["result"].get(field)
208
209			return cls(**newdoc)
210
211			@classmethod
212			def from_inquiry_api(cls, inquiry_api, mask_secrets=False):
213			""" Allows translation of InquiryAPI directly to InquiryResponseAPI
214
215			This bypasses the DB modeling, since there's no DB model for Inquiries yet.
216			"""
217
218			return cls(
219			id=getattr(inquiry_api, 'id', None),
220			route=getattr(inquiry_api, 'route', None),
221			ttl=getattr(inquiry_api, 'ttl', None),
222			users=getattr(inquiry_api, 'users', None),
223			roles=getattr(inquiry_api, 'roles', None),
224			schema=getattr(inquiry_api, 'schema', None)
225			)
226

StackStorm / st2

Push — master ( ac1734...fff02f )

InquiryAPI.from_model() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like