Commit e9fdbe08 authored by Luciano Cauzzi's avatar Luciano Cauzzi

Completed attachment mng

parent 23759d0f
......@@ -143,6 +143,34 @@ class AppPyramidManager:
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__()
try:
url = "{}/{}/_search".format(self.__base_url__, self.__res_url__)
headers = self.__get_auth_header__()
payload = "{\n\t\"_source\": {\n " \
"\"includes\": \"*\"},\n " \
"\n \"query\": {\n \"terms\": {\n \"" + query_header + "\": [ \"" \
+ query_id + "\"] \n }\n }\n}"
s = self.__network_manager__.get_session()
response = s.get(url, data=payload, headers=headers, timeout=AppPyramidManager.TIMEOUT)
if response.status_code == 401:
self.login()
return self.exists(query_header, query_id)
json_obj = json.loads(response.text)
attachments = json_obj["hits"]["hits"]
if response.status_code != 200:
return NetworkResult(success=False, status_code=response.status_code, message="", result=False)
if len(attachments) > 0:
return NetworkResult(success=True, status_code=response.status_code,
message=json_obj["hits"]["hits"][0]["_source"], result=True)
else:
return NetworkResult(success=True, status_code=response.status_code, message="", result=False)
except Exception as e:
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def __get_auth_header__(self):
if self.__app_token__ is None:
raise Exception("User should be authenticated")
......
......@@ -29,6 +29,11 @@ def from_list(f: Callable[[Any], T], x: Any) -> List[T]:
return [f(y) for y in x]
def from_bool(x: Any) -> bool:
assert isinstance(x, bool)
return x
@dataclass
class AppToken:
expire: int
......@@ -55,11 +60,13 @@ class File:
def from_dict(obj: Any) -> 'File':
assert isinstance(obj, dict)
name = from_str(obj.get("_name"))
content = from_str(obj.get("_content") if obj.get("_content") is not None else "")
return File(name=name, content=content)
content = from_str(obj.get("_content"))
return File(name, content)
def to_dict(self) -> dict:
result: dict = {"_name": from_str(self.name), "_content": from_str(self.content)}
result: dict = {}
result["_name"] = from_str(self.name)
result["_content"] = from_str(self.content)
return result
......@@ -85,8 +92,8 @@ class PyramidDocument:
@staticmethod
def from_dict(obj: Any) -> 'PyramidDocument':
assert isinstance(obj, dict)
# TODO: missing Attachment
return PyramidDocument([])
attachments = from_list(Attachment.from_dict, obj.get("attachments"))
return PyramidDocument(attachments)
def to_dict(self) -> dict:
result: dict = {"attachments": from_list(lambda x: to_class(Attachment, x), self.attachments)}
......
......@@ -3,7 +3,7 @@ import setuptools
setup(
name='network_manager',
version='0.1.12',
version='0.1.13',
packages=setuptools.find_packages(),
dependency_links=[],
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