Commit 58eb9719 authored by Luciano Cauzzi's avatar Luciano Cauzzi

added login check. refactor response type

parent 60ab1e1c
import json
from typing import Optional
from datetime import datetime
from logger_library import Logger
from network_manager.pyramid_model import AppToken, app_token_from_dict, PyramidDocument
from network_manager.pyramid_model import app_token_from_dict, PyramidDocument, NetworkResult, AppToken
logger = Logger.app_logger().get_logger()
......@@ -15,13 +16,14 @@ class AppPyramidManager:
self.__api_key__ = api_key
self.__base_url__ = base_url
self.__network_manager__ = network_manager
self.__app_token__ = None
self.__app_token__: AppToken = None
logger.debug(
"Initialization completed: \nBASE_URL= {} \nRESOURCE_URL= {}\nAPI_KEY= {}".format(base_url, resource_url,
api_key))
self.login()
def login(self) -> Optional[AppToken]:
def login(self) -> NetworkResult:
operation_name = "user login"
try:
url = "{}/sessions".format(self.__base_url__)
payload = "username=admin&api_key=" + self.__api_key__
......@@ -32,34 +34,50 @@ class AppPyramidManager:
response = s.post(url, data=payload, headers=headers, timeout=AppPyramidManager.TIMEOUT)
json_repr = json.loads(response.text)
self.__app_token__ = app_token_from_dict(json_repr)
logger.info("User has been authenticated")
return self.__app_token__
return AppPyramidManager.__manage_response__(response, operation_name, self.__app_token__)
except Exception as e:
logger.error("Error while authenticating into pyramid {} \n".format(e))
return None
return AppPyramidManager.__generate_generic_error__(operation_name, e)
@staticmethod
def __generate_generic_error__(operation_name: str, exception: Exception):
msg = "An exception occurred while {}: \n{} \n".format(operation_name, exception)
logger.error(msg)
logger.exception(exception)
return NetworkResult(success=False, status_code=400, message=msg, result=None)
def save(self, doc: PyramidDocument) -> bool:
@staticmethod
def __manage_response__(status_code: int, operation_name: str, result):
if status_code == 200:
logger.info(operation_name + ": SUCCEED!")
return NetworkResult(success=True, status_code=200, message="", result=result)
else:
logger.error(operation_name + ": FAILED!")
return NetworkResult(success=False, status_code=status_code, message="", result=None)
def save(self, doc: PyramidDocument) -> NetworkResult:
"""
Saves a document inside pyramid
:param doc:
:return: True or False
"""
operation_name = "saving document"
self.__perform_auth_if_needed__()
try:
url = "{}/{}".format(self.__base_url__, self.__res_url__)
headers = self.__get_auth_header__()
s = self.__network_manager__.get_session()
payload = json.dumps(doc.to_dict())
response = s.post(url, data=payload, headers=headers, timeout=AppPyramidManager.TIMEOUT)
if response.status_code != 200:
return False
if response.status_code == 201:
self.login()
return self.save(doc)
return AppPyramidManager.__manage_response__(response.status_code, operation_name, result=None)
except Exception as e:
logger.error("Error while uploading document to pyramid")
logger.exception(e)
return False
return True
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def exists(self, query_header, query_id) -> Optional[str]:
def exists(self, query_header, query_id) -> NetworkResult:
operation_name = "existence checking"
self.__perform_auth_if_needed__()
try:
url = "{}/{}/_search".format(self.__base_url__, self.__res_url__)
......@@ -71,19 +89,18 @@ class AppPyramidManager:
+ 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 == 201:
self.login()
return self.exists(query_header, query_id)
json_obj = json.loads(response.text)
attachments = json_obj["hits"]["hits"]
if len(attachments) > 0:
doc_id = json_obj["hits"]["hits"][0]["_id"]
return doc_id
else:
return None
return AppPyramidManager.__manage_response__(response.status_code, operation_name, json_obj["hits"]["hits"][0]["_id"] if len(attachments) > 0 else None)
except Exception as e:
logger.error("Error while checking document's existence")
logger.exception(e)
return None
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def update(self, elastic_id: str, doc: PyramidDocument) -> bool:
def update(self, elastic_id: str, doc: PyramidDocument) -> NetworkResult:
operation_name = "saving document"
self.__perform_auth_if_needed__()
try:
url = "{}/{}/{}".format(self.__base_url__, self.__res_url__, elastic_id)
......@@ -91,13 +108,12 @@ class AppPyramidManager:
s = self.__network_manager__.get_session()
payload = json.dumps(doc.to_dict())
response = s.put(url, data=payload, headers=headers, timeout=AppPyramidManager.TIMEOUT)
if response.status_code != 200:
return False
if response.status_code == 201:
self.login()
return self.update(elastic_id, doc)
return AppPyramidManager.__manage_response__(response.status_code, operation_name, None)
except Exception as e:
logger.error("Error while uploading injunctions to server")
logger.exception(e)
return False
return True
return AppPyramidManager.__generate_generic_error__(operation_name, e)
def __get_auth_header__(self):
if self.__app_token__ is None:
......@@ -106,3 +122,15 @@ class AppPyramidManager:
'AUTHORIZATION': self.__app_token__.token,
'Content-Type': "application/json"
}
def __perform_auth_if_needed__(self):
if self.__is_token_expired__():
self.login()
def __is_token_expired__(self) -> bool:
return self.__app_token__.expire < AppPyramidManager.__unix_time_millis__(datetime.now())
@staticmethod
def __unix_time_millis__(dt):
epoch = datetime.utcfromtimestamp(0)
return (dt - epoch).total_seconds() * 1000.0
......@@ -93,5 +93,13 @@ class PyramidDocument:
return result
@dataclass
class NetworkResult:
success: bool
status_code: Optional[int]
message: Optional[str]
result: Optional[T]
def app_token_from_dict(s: Any) -> AppToken:
return AppToken.from_dict(s)
......@@ -3,7 +3,7 @@ import setuptools
setup(
name='network_manager',
version='0.1.2',
version='0.1.3',
packages=setuptools.find_packages(),
dependency_links=['http://github.comune.brescia.it/innovation-that-excites/logger_library#egg=logger_library'],
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