tabpy.tabpy_server.psws.python_service.PythonServiceHandler.manage_request() - Code Metrics - tableau/TabPy - Measure and Improve Code Quality continuously with Scrutinizer

PythonServiceHandler.manage_request() B
last analyzed 2024-11-25 22:00 UTC

↳ Parent: tabpy.tabpy_server.psws.python_service

Complexity

Conditions

Size

Total Lines	25
Code Lines	23

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	64.006

Importance

Changes

Metric	Value
eloc	23
dl	0
loc	25
ccs	1
cts	23
cp	0.0434
rs	7.3333
c	0
b	0
f	0
cc	8
nop	2
crap	64.006

import concurrent.futures
import logging
from tabpy.tabpy_server.common.util import format_exception
from tabpy.tabpy_server.common.messages import (
    LoadObject,
    DeleteObjects,
    FlushObjects,
    CountObjects,
    ListObjects,
    UnknownMessage,
    LoadFailed,
    ObjectsDeleted,
    ObjectsFlushed,
    QueryFailed,
    QuerySuccessful,
    UnknownURI,
    DownloadSkipped,
    LoadInProgress,
    ObjectCount,
    ObjectList,
)
from tabpy.tabpy_tools.query_object import QueryObject


logger = logging.getLogger(__name__)


class PythonServiceHandler:
    """
    A wrapper around PythonService object that receives requests and calls the
    corresponding methods.
    """

    def __init__(self, ps):
        self.ps = ps

    def manage_request(self, msg):
        try:
            logger.debug(f"Received request {type(msg).__name__}")
            if isinstance(msg, LoadObject):
                response = self.ps.load_object(*msg)
            elif isinstance(msg, DeleteObjects):
                response = self.ps.delete_objects(msg.uris)
            elif isinstance(msg, FlushObjects):
                response = self.ps.flush_objects()
            elif isinstance(msg, CountObjects):
                response = self.ps.count_objects()
            elif isinstance(msg, ListObjects):
                response = self.ps.list_objects()
            else:
                response = UnknownMessage(msg)

            logger.debug(f"Returning response {response}")
            return response
        except Exception as e:
            logger.exception(e)
            msg = e
            if hasattr(e, "message"):
                msg = e.message
            logger.error(f"Error processing request: {msg}")
            return UnknownMessage(msg)


class PythonService:
    """
    This class is a simple wrapper maintaining loaded query objects from
    the current TabPy instance. `query_objects` is a dictionary that
    maps query object URI to query objects

    The query_objects schema is as follow:

    {'version': <current-successfuly-loaded-version>,
     'last_error':<your-recent-error-to-load-model>,
     'endpoint_obj':<loaded_query_objects>,
     'type':<object-type>,
     'status':<LoadSuccessful-or-LoadFailed-or-LoadInProgress>}

    """

    def __init__(self, query_objects=None):

        self.EXECUTOR = concurrent.futures.ThreadPoolExecutor(max_workers=1)
        self.query_objects = query_objects or {}

    def _load_object(
        self, object_uri, object_url, object_version, is_update, object_type
    ):
        try:
            logger.info(
                f"Loading object:, URI={object_uri}, "
                f"URL={object_url}, version={object_version}, "
                f"is_updated={is_update}"
            )
            if object_type == "model":
                po = QueryObject.load(object_url)
            elif object_type == "alias":
                po = object_url
            else:
                raise RuntimeError(f"Unknown object type: {object_type}")

            self.query_objects[object_uri] = {
                "version": object_version,
                "type": object_type,
                "endpoint_obj": po,
                "status": "LoadSuccessful",
                "last_error": None,
            }
        except Exception as e:
            logger.exception(e)
            logger.error(
                f"Unable to load QueryObject: path={object_url}, " f"error={str(e)}"
            )

            self.query_objects[object_uri] = {
                "version": object_version,
                "type": object_type,
                "endpoint_obj": None,
                "status": "LoadFailed",
                "last_error": f"Load failed: {str(e)}",
            }

    def load_object(
        self, object_uri, object_url, object_version, is_update, object_type
    ):
        try:
            obj_info = self.query_objects.get(object_uri)
            if (
                obj_info
                and obj_info["endpoint_obj"]
                and (obj_info["version"] >= object_version)
            ):
                logger.info("Received load message for object already loaded")

                return DownloadSkipped(
                    object_uri,
                    obj_info["version"],
                    "Object with greater " "or equal version already loaded",
                )
            else:
                if object_uri not in self.query_objects:
                    self.query_objects[object_uri] = {
                        "version": object_version,
                        "type": object_type,
                        "endpoint_obj": None,
                        "status": "LoadInProgress",
                        "last_error": None,
                    }
                else:
                    self.query_objects[object_uri]["status"] = "LoadInProgress"

                self.EXECUTOR.submit(
                    self._load_object,
                    object_uri,
                    object_url,
                    object_version,
                    is_update,
                    object_type,
                )

                return LoadInProgress(
                    object_uri, object_url, object_version, is_update, object_type
                )
        except Exception as e:
            logger.exception(e)
            logger.error(
                f"Unable to load QueryObject: path={object_url}, " f"error={str(e)}"
            )

            self.query_objects[object_uri] = {
                "version": object_version,
                "type": object_type,
                "endpoint_obj": None,
                "status": "LoadFailed",
                "last_error": str(e),
            }

            return LoadFailed(object_uri, object_version, str(e))

    def delete_objects(self, object_uris):
        """Delete one or more objects from the query_objects map"""
        if isinstance(object_uris, list):
            deleted = []
            for uri in object_uris:
                deleted.extend(self.delete_objects(uri).uris)
            return ObjectsDeleted(deleted)
        elif isinstance(object_uris, str):
            deleted_obj = self.query_objects.pop(object_uris, None)
            if deleted_obj:
                return ObjectsDeleted([object_uris])
            else:
                logger.warning(
                    f"Received message to delete query object "
                    f"that doesn't exist: "
                    f"object_uris={object_uris}"
                )
                return ObjectsDeleted([])
        else:
            logger.error(
                f"Unexpected input to delete objects: input={object_uris}, "
                f'info="Input should be list or str. '
                f'Type: {type(object_uris)}"'
            )
            return ObjectsDeleted([])

    def flush_objects(self):
        """Flush objects from the query_objects map"""
        logger.debug("Flushing query objects")
        n = len(self.query_objects)
        self.query_objects.clear()
        return ObjectsFlushed(n, 0)

    def count_objects(self):
        """Count the number of Loaded QueryObjects stored in memory"""
        count = 0
        for uri, po in self.query_objects.items():
            if po["endpoint_obj"] is not None:
                count += 1
        return ObjectCount(count)

    def list_objects(self):
        """List the objects as (URI, version) pairs"""

        objects = {}
        for (uri, obj_info) in self.query_objects.items():
            objects[uri] = {
                "version": obj_info["version"],
                "type": obj_info["type"],
                "status": obj_info["status"],
                "reason": obj_info["last_error"],
            }

        return ObjectList(objects)

    def query(self, object_uri, params, uid):
        """Execute a QueryObject query"""
        logger.debug(f"Querying Python service {object_uri}...")
        try:
            if not isinstance(params, dict) and not isinstance(params, list):
                return QueryFailed(
                    uri=object_uri,
                    error=(
                        "Query parameter needs to be a dictionary or a list"
                        f". Given value is of type {type(params)}"
                    ),
                )

            obj_info = self.query_objects.get(object_uri)
            logger.debug(f"Found object {obj_info}")
            if obj_info:
                pred_obj = obj_info["endpoint_obj"]
                version = obj_info["version"]

                if not pred_obj:
                    return QueryFailed(
                        uri=object_uri,
                        error=(
                            "There is no query object associated to the "
                            f"endpoint: {object_uri}"
                        ),
                    )

                logger.debug(f"Querying endpoint with params ({params})...")
                if isinstance(params, dict):
                    result = pred_obj.query(**params)
                else:
                    result = pred_obj.query(*params)

                return QuerySuccessful(object_uri, version, result)
            else:
                return UnknownURI(object_uri)
        except Exception as e:
            logger.exception(e)
            err_msg = format_exception(e, "/query")
            logger.error(err_msg)
            return QueryFailed(uri=object_uri, error=err_msg)


1	1	import concurrent.futures
2	1	import logging
3	1	from tabpy.tabpy_server.common.util import format_exception
4	1	from tabpy.tabpy_server.common.messages import (
5		LoadObject,
6		DeleteObjects,
7		FlushObjects,
8		CountObjects,
9		ListObjects,
10		UnknownMessage,
11		LoadFailed,
12		ObjectsDeleted,
13		ObjectsFlushed,
14		QueryFailed,
15		QuerySuccessful,
16		UnknownURI,
17		DownloadSkipped,
18		LoadInProgress,
19		ObjectCount,
20		ObjectList,
21		)
22	1	from tabpy.tabpy_tools.query_object import QueryObject
23
24
25	1	logger = logging.getLogger(__name__)
26
27
28	1	class PythonServiceHandler:
29		"""
30		A wrapper around PythonService object that receives requests and calls the
31		corresponding methods.
32		"""
33
34	1	def __init__(self, ps):
35	1	self.ps = ps
36
37	1	def manage_request(self, msg):
38		try:
39		logger.debug(f"Received request {type(msg).__name__}")
40		if isinstance(msg, LoadObject):
41		response = self.ps.load_object(*msg)
42		elif isinstance(msg, DeleteObjects):
43		response = self.ps.delete_objects(msg.uris)
44		elif isinstance(msg, FlushObjects):
45		response = self.ps.flush_objects()
46		elif isinstance(msg, CountObjects):
47		response = self.ps.count_objects()
48		elif isinstance(msg, ListObjects):
49		response = self.ps.list_objects()
50		else:
51		response = UnknownMessage(msg)
52
53		logger.debug(f"Returning response {response}")
54		return response
55		except Exception as e:
56		logger.exception(e)
57		msg = e
58		if hasattr(e, "message"):
59		msg = e.message
60		logger.error(f"Error processing request: {msg}")
61		return UnknownMessage(msg)
62
63
64	1	class PythonService:
65		"""
66		This class is a simple wrapper maintaining loaded query objects from
67		the current TabPy instance. `query_objects` is a dictionary that
68		maps query object URI to query objects
69
70		The query_objects schema is as follow:
71
72		{'version': <current-successfuly-loaded-version>,
73		'last_error':<your-recent-error-to-load-model>,
74		'endpoint_obj':<loaded_query_objects>,
75		'type':<object-type>,
76		'status':<LoadSuccessful-or-LoadFailed-or-LoadInProgress>}
77
78		"""
79
80	1	def __init__(self, query_objects=None):
81
82	1	self.EXECUTOR = concurrent.futures.ThreadPoolExecutor(max_workers=1)
83	1	self.query_objects = query_objects or {}
84
85	1	def _load_object(
86		self, object_uri, object_url, object_version, is_update, object_type
87		):
88		try:
89		logger.info(
90		f"Loading object:, URI={object_uri}, "
91		f"URL={object_url}, version={object_version}, "
92		f"is_updated={is_update}"
93		)
94		if object_type == "model":
95		po = QueryObject.load(object_url)
96		elif object_type == "alias":
97		po = object_url
98		else:
99		raise RuntimeError(f"Unknown object type: {object_type}")
100
101		self.query_objects[object_uri] = {
102		"version": object_version,
103		"type": object_type,
104		"endpoint_obj": po,
105		"status": "LoadSuccessful",
106		"last_error": None,
107		}
108		except Exception as e:
109		logger.exception(e)
110		logger.error(
111		f"Unable to load QueryObject: path={object_url}, " f"error={str(e)}"
112		)
113
114		self.query_objects[object_uri] = {
115		"version": object_version,
116		"type": object_type,
117		"endpoint_obj": None,
118		"status": "LoadFailed",
119		"last_error": f"Load failed: {str(e)}",
120		}
121
122	1	def load_object(
123		self, object_uri, object_url, object_version, is_update, object_type
124		):
125		try:
126		obj_info = self.query_objects.get(object_uri)
127		if (
128		obj_info
129		and obj_info["endpoint_obj"]
130		and (obj_info["version"] >= object_version)
131		):
132		logger.info("Received load message for object already loaded")
133
134		return DownloadSkipped(
135		object_uri,
136		obj_info["version"],
137		"Object with greater " "or equal version already loaded",
138		)
139		else:
140		if object_uri not in self.query_objects:
141		self.query_objects[object_uri] = {
142		"version": object_version,
143		"type": object_type,
144		"endpoint_obj": None,
145		"status": "LoadInProgress",
146		"last_error": None,
147		}
148		else:
149		self.query_objects[object_uri]["status"] = "LoadInProgress"
150
151		self.EXECUTOR.submit(
152		self._load_object,
153		object_uri,
154		object_url,
155		object_version,
156		is_update,
157		object_type,
158		)
159
160		return LoadInProgress(
161		object_uri, object_url, object_version, is_update, object_type
162		)
163		except Exception as e:
164		logger.exception(e)
165		logger.error(
166		f"Unable to load QueryObject: path={object_url}, " f"error={str(e)}"
167		)
168
169		self.query_objects[object_uri] = {
170		"version": object_version,
171		"type": object_type,
172		"endpoint_obj": None,
173		"status": "LoadFailed",
174		"last_error": str(e),
175		}
176
177		return LoadFailed(object_uri, object_version, str(e))
178
179	1	def delete_objects(self, object_uris):
180		"""Delete one or more objects from the query_objects map"""
181		if isinstance(object_uris, list):
182		deleted = []
183		for uri in object_uris:
184		deleted.extend(self.delete_objects(uri).uris)
185		return ObjectsDeleted(deleted)
186		elif isinstance(object_uris, str):
187		deleted_obj = self.query_objects.pop(object_uris, None)
188		if deleted_obj:
189		return ObjectsDeleted([object_uris])
190		else:
191		logger.warning(
192		f"Received message to delete query object "
193		f"that doesn't exist: "
194		f"object_uris={object_uris}"
195		)
196		return ObjectsDeleted([])
197		else:
198		logger.error(
199		f"Unexpected input to delete objects: input={object_uris}, "
200		f'info="Input should be list or str. '
201		f'Type: {type(object_uris)}"'
202		)
203		return ObjectsDeleted([])
204
205	1	def flush_objects(self):
206		"""Flush objects from the query_objects map"""
207		logger.debug("Flushing query objects")
208		n = len(self.query_objects)
209		self.query_objects.clear()
210		return ObjectsFlushed(n, 0)
211
212	1	def count_objects(self):
213		"""Count the number of Loaded QueryObjects stored in memory"""
214		count = 0
215		for uri, po in self.query_objects.items():
216		if po["endpoint_obj"] is not None:
217		count += 1
218		return ObjectCount(count)
219
220	1	def list_objects(self):
221		"""List the objects as (URI, version) pairs"""
222
223		objects = {}
224		for (uri, obj_info) in self.query_objects.items():
225		objects[uri] = {
226		"version": obj_info["version"],
227		"type": obj_info["type"],
228		"status": obj_info["status"],
229		"reason": obj_info["last_error"],
230		}
231
232		return ObjectList(objects)
233
234	1	def query(self, object_uri, params, uid):
235		"""Execute a QueryObject query"""
236		logger.debug(f"Querying Python service {object_uri}...")
237		try:
238		if not isinstance(params, dict) and not isinstance(params, list):
239		return QueryFailed(
240		uri=object_uri,
241		error=(
242		"Query parameter needs to be a dictionary or a list"
243		f". Given value is of type {type(params)}"
244		),
245		)
246
247		obj_info = self.query_objects.get(object_uri)
248		logger.debug(f"Found object {obj_info}")
249		if obj_info:
250		pred_obj = obj_info["endpoint_obj"]
251		version = obj_info["version"]
252
253		if not pred_obj:
254		return QueryFailed(
255		uri=object_uri,
256		error=(
257		"There is no query object associated to the "
258		f"endpoint: {object_uri}"
259		),
260		)
261
262		logger.debug(f"Querying endpoint with params ({params})...")
263		if isinstance(params, dict):
264		result = pred_obj.query(**params)
265		else:
266		result = pred_obj.query(*params)
267
268		return QuerySuccessful(object_uri, version, result)
269		else:
270		return UnknownURI(object_uri)
271		except Exception as e:
272		logger.exception(e)
273		err_msg = format_exception(e, "/query")
274		logger.error(err_msg)
275		return QueryFailed(uri=object_uri, error=err_msg)
276

tableau / TabPy

PythonServiceHandler.manage_request() B last analyzed 2024-11-25 22:00 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

PythonServiceHandler.manage_request() B
last analyzed 2024-11-25 22:00 UTC