Commit f4854d5e authored by Luciano Cauzzi's avatar Luciano Cauzzi

Added pagination function

parent b8a52bcd
......@@ -9,9 +9,11 @@ class NetworkManager(containers.DeclarativeContainer):
class PyramidManager(containers.DeclarativeContainer):
PAGINATION_SIZE = 20
config = providers.Configuration('config')
manager = providers.Factory(AppPyramidManager,
base_url=config.pyramid.base_url,
resource_url=config.pyramid.res_url,
api_key=config.pyramid.api_key,
network_manager=NetworkManager.manager())
network_manager=NetworkManager.manager(),
pagination_size=PAGINATION_SIZE)
......@@ -11,12 +11,13 @@ logger = Logger.app_logger().get_logger()
class AppPyramidManager:
TIMEOUT = 30
def __init__(self, base_url, resource_url, api_key, network_manager):
def __init__(self, base_url, resource_url, api_key, network_manager, pagination_size):
self.__res_url__ = resource_url
self.__api_key__ = api_key
self.__base_url__ = base_url
self.__network_manager__ = network_manager
self.__app_token__: AppToken = None
self.__pagination_size__ = pagination_size
logger.debug(
"Initialization completed: \nBASE_URL= {} \nRESOURCE_URL= {}\nAPI_KEY= {}".format(base_url, resource_url,
api_key))
......@@ -143,6 +144,27 @@ 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):
operation_name = "get paginated"
self.__perform_auth_if_needed__()
try:
url = "{}/{}/_search".format(self.__base_url__, self.__res_url__)
payload = {
"size": size,
"_source": excludes,
"from": element_from,
"query": query,
"aggs": aggs
}
headers = self.__get_auth_header__()
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)
return object_repr
except Exception as e:
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def get_attachments(self, query_header, query_id) -> NetworkResult:
operation_name = "get_attachments"
self.__perform_auth_if_needed__()
......
......@@ -4,6 +4,20 @@ from typing import Any, Optional, TypeVar, cast, Type, Callable, List
T = TypeVar("T")
def from_none(x: Any) -> Any:
assert x is None
return x
def from_union(fs, x):
for f in fs:
try:
return f(x)
except:
pass
assert False
def from_str(x: Any) -> str:
assert isinstance(x, str)
return x
......@@ -34,6 +48,11 @@ def from_bool(x: Any) -> bool:
return x
def from_float(x: Any) -> float:
assert isinstance(x, (float, int)) and not isinstance(x, bool)
return float(x)
@dataclass
class AppToken:
expire: int
......@@ -64,9 +83,7 @@ class File:
return File(name, content)
def to_dict(self) -> dict:
result: dict = {}
result["_name"] = from_str(self.name)
result["_content"] = from_str(self.content)
result: dict = {"_name": from_str(self.name), "_content": from_str(self.content)}
return result
......
......@@ -3,7 +3,7 @@ import setuptools
setup(
name='network_manager',
version='0.1.14',
version='0.1.15',
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