Integrated login function

network_manager/app_network_manager.py

@@ -21,3 +21,6 @@ class AppNetworkManager:
         self.__session.mount('http://', adapter)
         self.__session.mount('https://', adapter)
         pass
+
+    def get_session(self):
+        return self.__session

network_manager/container.py

@@ -10,4 +10,8 @@ class NetworkManager(containers.DeclarativeContainer):
 
 class PyramidManager(containers.DeclarativeContainer):
     config = providers.Configuration('config')
-    manager = providers.Factory(AppPyramidManager, base_url=config.pyramid.base_url, resource_url=config.pyramid.res_url,  token=config.pyramid.token)
+    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/pyramid_manager.py

+import json
+from typing import Optional
+
 from logger_library import Logger
-from .pyramid_model import *
+from network_manager.pyramid_model import AppToken, app_token_from_dict
 
 logger = Logger.app_logger().get_logger()
 
 
 class AppPyramidManager:
+    TIMEOUT = 30
 
-    def __init__(self, base_url, resource_url, token):
+    def __init__(self, base_url, resource_url, api_key, network_manager):
         self.__url__ = resource_url
-        self.__token__ = token
+        self.__api_key__ = api_key
         self.__base_url__ = base_url
-        logger.debug("Initialization completed: \nBASE_URL= {} \nRESOURCE_URL=   {}\nTOKEN=  {}".format(base_url, resource_url, token))
+        self.__network_manager__ = network_manager
+        logger.debug("Initialization completed: \nBASE_URL= {} \nRESOURCE_URL=   {}\nAPI_KEY=  {}".format(base_url, resource_url, api_key))
+
+    def login(self) -> Optional[AppToken]:
+        try:
+            url = "{}/sessions".format(self.__base_url__)
+            payload = "username=admin&api_key=" + self.__api_key__
+            headers = {
+                'Content-Type': "application/x-www-form-urlencoded"
+            }
+            s = self.__network_manager__.get_session()
+            response = s.post(url, data=payload, headers=headers, timeout=AppPyramidManager.TIMEOUT)
+            json_repr = json.loads(response.text)
+            return app_token_from_dict(json_repr)
+        except Exception as e:
+            logger.error("\n[NETWORK]: Error while fetching police report list: {} \n".format(e))
+        return None

network_manager/pyramid_model.py

@@ -44,3 +44,7 @@ class AppToken:
     def to_dict(self) -> dict:
         result: dict = {"expire": from_int(self.expire), "token": from_str(self.token)}
         return result
+
+
+def app_token_from_dict(s: Any) -> AppToken:
+    return AppToken.from_dict(s)

run.py

@@ -12,10 +12,11 @@ def main():
         'pyramid': {
             'base_url': BASE_URL,
             'res_url': RES_URL,
-            'token': TOKEN,
+            'api_key': TOKEN,
         }
     }
     manager = PyramidManager(config=config).manager()
+    login = manager.login()
     pass