Commit 1f90d78d authored by Luciano Cauzzi's avatar Luciano Cauzzi

Move search model inside library

parent f4854d5e
......@@ -3,7 +3,7 @@ from datetime import datetime, timezone
from logger_library import Logger
from network_manager.pyramid_model import app_token_from_dict, PyramidDocument, NetworkResult, AppToken
from network_manager.pyramid_model import app_token_from_dict, PyramidDocument, NetworkResult, AppToken, ResultList
logger = Logger.app_logger().get_logger()
......@@ -144,7 +144,7 @@ class AppPyramidManager:
except Exception as e:
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def get_paginated(self, return_class_type, element_from: int, query: dict, excludes: dict, aggs: dict, size: int = 20):
def get_paginated(self, item_class_type, element_from: int, query: dict, excludes: dict, aggs: dict, size: int = 20):
operation_name = "get paginated"
self.__perform_auth_if_needed__()
try:
......@@ -160,7 +160,7 @@ class AppPyramidManager:
s = self.__network_manager__.get_session()
response = s.get(url, headers=headers, data=json.dumps(payload), timeout=AppPyramidManager.TIMEOUT)
json_repr = json.loads(response.text)
object_repr = return_class_type.from_dict(json_repr)
object_repr = ResultList.from_dict(json_repr, item_class_type)
return object_repr
except Exception as e:
return AppPyramidManager.__generate_generic_error__(operation_name, e)
......
......@@ -4,6 +4,20 @@ from typing import Any, Optional, TypeVar, cast, Type, Callable, List
T = TypeVar("T")
def from_list_with_class(f: Callable[[Any, Any], T], x: Any, class_type: T) -> List[T]:
assert isinstance(x, list)
return [f(y, class_type) for y in x]
def from_union_with_class(fs, x, class_type: T):
for f in fs:
try:
return f(x, class_type)
except:
pass
assert False
def from_none(x: Any) -> Any:
assert x is None
return x
......@@ -117,6 +131,93 @@ class PyramidDocument:
return result
@dataclass
class Hit:
id: Optional[str] = None
score: Optional[int] = None
source: Optional[Any] = None
@staticmethod
def from_dict(obj: Any, class_type: T) -> 'Hit':
assert isinstance(obj, dict)
id = from_union([from_str, from_none], obj.get("_id"))
score = from_union([from_float, from_none], obj.get("_score"))
source = from_union([class_type.from_dict, from_none], obj.get("_source"))
return Hit(id, score, source)
def to_dict(self, class_type: T) -> dict:
result: dict = {"_id": from_union([from_str, from_none], self.id),
"_score": from_union([from_float, from_none], self.score),
"_source": from_union([lambda x: to_class(class_type, x), from_none], self.source)}
return result
@dataclass
class Hits:
total: Optional[int] = None
max_score: Optional[int] = None
hits: Optional[List[Hit]] = None
@staticmethod
def from_dict(obj: Any, class_type: T) -> 'Hits':
assert isinstance(obj, dict)
total = from_union([from_int, from_none], obj.get("total"))
max_score = from_union([from_float, from_none], obj.get("max_score"))
hits = from_union([lambda x: from_list_with_class(Hit.from_dict, x, class_type), from_none], obj.get("hits"))
return Hits(total, max_score, hits)
def to_dict(self) -> dict:
result: dict = {"total": from_union([from_int, from_none], self.total),
"max_score": from_union([from_float, from_none], self.max_score),
"hits": from_union([lambda x: from_list(lambda x: to_class(Hit, x), x), from_none], self.hits)}
return result
@dataclass
class Shards:
total: Optional[int] = None
successful: Optional[int] = None
failed: Optional[int] = None
@staticmethod
def from_dict(obj: Any) -> 'Shards':
assert isinstance(obj, dict)
total = from_union([from_int, from_none], obj.get("total"))
successful = from_union([from_int, from_none], obj.get("successful"))
failed = from_union([from_int, from_none], obj.get("failed"))
return Shards(total, successful, failed)
def to_dict(self) -> dict:
result: dict = {"total": from_union([from_int, from_none], self.total),
"successful": from_union([from_int, from_none], self.successful),
"failed": from_union([from_int, from_none], self.failed)}
return result
@dataclass
class ResultList:
took: Optional[int] = None
timed_out: Optional[bool] = None
shards: Optional[Shards] = None
hits: Optional[Hits] = None
@staticmethod
def from_dict(obj: Any, class_type: T) -> 'ResultList':
assert isinstance(obj, dict)
took = from_union([from_int, from_none], obj.get("took"))
timed_out = from_union([from_bool, from_none], obj.get("timed_out"))
shards = from_union([Shards.from_dict, from_none], obj.get("_shards"))
hits = from_union_with_class([Hits.from_dict, from_none], obj.get("hits"), class_type)
return ResultList(took, timed_out, shards, hits)
def to_dict(self) -> dict:
result: dict = {"took": from_union([from_int, from_none], self.took),
"timed_out": from_union([from_bool, from_none], self.timed_out),
"_shards": from_union([lambda x: to_class(Shards, x), from_none], self.shards),
"hits": from_union([lambda x: to_class(Hits, x), from_none], self.hits)}
return result
@dataclass
class NetworkResult:
success: bool
......
......@@ -3,7 +3,7 @@ import setuptools
setup(
name='network_manager',
version='0.1.15',
version='0.1.16',
packages=setuptools.find_packages(),
dependency_links=['http://github.comune.brescia.it/innovation-that-excites/logger_library#egg=logger_library-0.1.18'],
install_requires=['logger_library', 'dependency_injector', 'requests'],
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment