Merge pull request #14 from arrowhead-f/decouple-consumer

Decouple consumer

Author: ajoino

arrowhead_client/__init__.py

@@ -1,17 +0,0 @@
-#!/usr/bin/env python
-
-from . import configuration
-from .system import ArrowheadSystem
-from .system import ConsumerSystem
-from .system import ProviderSystem
-from .service import ConsumedHttpService as ConsumedService
-from .service import ProvidedHttpService as ProvidedService
-
-config = configuration.set_config()
-
-#TODO: This line should be removed and the certificates should be added
-# to the list of trusted certificates.
-import requests
-from requests.packages.urllib3.exceptions import InsecureRequestWarning
-requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
-

arrowhead_client/api.py

@@ -0,0 +1,28 @@
+from arrowhead_client.configuration import config
+from arrowhead_client.application import ArrowheadApplication
+from arrowhead_client.system import ArrowheadSystem
+from arrowhead_client.consumer import Consumer
+from arrowhead_client.provider import Provider
+from arrowhead_client.logs import get_logger
+
+
+
+class ArrowheadHttpApplication(ArrowheadApplication):
+    def __init__(self,
+                 system_name: str,
+                 address: str,
+                 port: int,
+                 authentication_info: str = '',
+                 keyfile: str = '',
+                 certfile: str = ''):
+        super().__init__(
+                ArrowheadSystem(system_name, address, port, authentication_info),
+                Consumer(keyfile, certfile),
+                Provider(),
+                get_logger(system_name, 'debug'),
+                config,
+                keyfile=keyfile,
+                certfile=certfile
+        )
+        self._logger.info(f'{self.__class__.__name__} initialized at {self.system.address}:{self.system.port}')
+        #TODO: This line is a hack and needs to be fixed

arrowhead_client/application.py

@@ -0,0 +1,224 @@
+from __future__ import annotations
+from typing import Any, Dict, Tuple
+from gevent import pywsgi # type: ignore
+from arrowhead_client.consumer import Consumer
+from arrowhead_client.provider import Provider
+from arrowhead_client.service import Service
+from arrowhead_client.core_services import core_service
+from arrowhead_client.system import ArrowheadSystem
+from arrowhead_client import core_service_forms as forms
+from arrowhead_client import core_service_responses as responses
+
+
+class ArrowheadApplication():
+    def __init__(self,
+                 system: ArrowheadSystem,
+                 consumer: Consumer,
+                 provider: Provider,
+                 logger: Any,
+                 config: Dict,
+                 server: Any = None,  # type: ignore
+                 keyfile: str = '',
+                 certfile: str  = '', ):
+        self.system = system
+        self.consumer = consumer
+        self.provider = provider
+        self._logger = logger
+        self.keyfile = keyfile
+        self.certfile = certfile
+        self.config = config
+        #TODO: Remove this hardcodedness
+        self.server = pywsgi.WSGIServer((self.system.address, self.system.port), self.provider.app,
+                                        keyfile=self.keyfile, certfile=self.certfile,
+                                        log=self._logger)
+        self._core_system_setup()
+        self.add_provided_service = self.provider.add_provided_service
+
+    @property
+    def cert(self) -> Tuple[str, str]:
+        return self.certfile, self.keyfile
+
+    '''
+    @classmethod
+    def from_cfg(cls, properties_file: str) -> ArrowheadHttpApplication:
+        """ Creates a BaseArrowheadSystem from a descriptor file """
+
+        # Parse configuration file
+        config = configparser.ConfigParser()
+        with open(properties_file, 'r') as properties:
+            config.read_file(properties)
+        config_dict = {k: v for k, v in config.items('SYSTEM')}
+
+        # Create class instance
+        system = cls(**config_dict)
+
+        return system
+    '''
+
+    def _core_system_setup(self) -> None:
+        service_registry = ArrowheadSystem(
+                'service_registry',
+                str(self.config['service_registry']['address']),
+                int(self.config['service_registry']['port']),
+                ''
+        )
+        orchestrator = ArrowheadSystem(
+                'orchestrator',
+                str(self.config['orchestrator']['address']),
+                int(self.config['orchestrator']['port']),
+                '')
+
+        self._store_consumed_service(core_service('register'), service_registry, 'POST')
+        self._store_consumed_service(core_service('unregister'), service_registry, 'DELETE')
+        self._store_consumed_service(core_service('orchestration-service'), orchestrator, 'POST')
+
+    def consume_service(self, service_definition: str, **kwargs):
+        return self.consumer.consume_service(service_definition, **kwargs)
+
+    def add_consumed_service(self,
+                             service_definition: str,
+                             http_method: str) -> None:
+        """ Add orchestration rule for service definition """
+
+        orchestration_form = forms.OrchestrationForm(self.system.dto, service_definition)
+
+        orchestration_response = self.consume_service('orchestration-service',
+                                                      json=orchestration_form.dto,
+                                                      cert=self.cert,
+                                                      )
+        #TODO: Handle orchestrator error codes
+
+        orchestration_payload = orchestration_response.json() # This might change with backend
+
+        (orchestrated_service, system), *_ = responses.handle_orchestration_response(orchestration_payload)
+
+        #TODO: Handle response with more than 1 service
+        # Perhaps a list of consumed services for each service definition should be stored
+        self._store_consumed_service(orchestrated_service, system, http_method)
+
+    def _store_consumed_service(self,
+                                service: Service,
+                                system: ArrowheadSystem,
+                                http_method: str) -> None:
+        """ Register consumed services with the consumer """
+
+        self.consumer._consumed_services[service.service_definition] = (service, system, http_method)
+
+    def provided_service(self,
+                         service_definition: str,
+                         service_uri: str,
+                         interface: str,
+                         method: str):
+        def wrapped_func(func):
+            self.provider.add_provided_service(
+                    service_definition,
+                    service_uri,
+                    interface,
+                    http_method=method,
+                    view_func=func)
+            return func
+        return wrapped_func
+
+    def _register_service(self, service: Service):
+        """ Registers the given service with service registry """
+
+        service_registration_form = forms.ServiceRegistrationForm(
+                service_definition=service.service_definition,
+                service_uri=service.service_uri,
+                secure='CERTIFICATE',
+                # TODO: secure should _NOT_ be hardcoded
+                interfaces=service.interface.dto,
+                provider_system=self.system.dto
+        )
+
+        service_registration_response = self.consume_service(
+                'register',
+                json=service_registration_form.dto,
+        )
+
+        print(service_registration_response.status_code)
+        # TODO: Error handling
+
+        # TODO: Do logging
+
+
+    def _register_all_services(self) -> None:
+        """ Registers all services of the system. """
+        for service, _ in self.provider.provided_services.values():
+            self._register_service(service)
+
+
+    def _unregister_service(self, service_definition: str) -> None:
+        """ Unregisters the given service with the service registry. """
+
+        if service_definition not in self.provider.provided_services.keys():
+            raise ValueError(f'{service_definition} not provided by {self}')
+
+        unregistration_payload = {
+            'service_definition': service_definition,
+            'system_name': self.system.system_name,
+            'address': self.system.address,
+            'port': self.system.port
+        }
+
+        service_unregistration_response = self.consume_service(
+                'unregister',
+                params=unregistration_payload
+        )
+
+        print(service_unregistration_response.status_code)
+
+
+    def _unregister_all_services(self) -> None:
+        """ Unregisters all services of the system """
+
+        for service_definition in self.provider.provided_services:
+            self._unregister_service(service_definition)
+
+
+    def run_forever(self) -> None:
+        """ Start the server, publish all service, and run until interrupted. Then, unregister all services"""
+
+        import warnings
+        warnings.simplefilter('ignore')
+
+        self._register_all_services()
+        try:
+            self._logger.info(f'Starting server')
+            print('Started Arrowhead System')
+            self.server.serve_forever()
+        except KeyboardInterrupt:
+            self._logger.info(f'Shutting down server')
+            print('Shutting down Arrowhead system')
+            self._unregister_all_services()
+        finally:
+            self._logger.info(f'Server shut down')
+
+
+    """
+    def __enter__(self):
+        '''Start server and register all services'''
+        import warnings
+        warnings.simplefilter('ignore')
+
+        print('Starting server')
+        self.server.start()
+        print('Registering services')
+        self.register_all_services()
+
+    def __exit__(self, exc_type, exc_value, tb):
+        '''Unregister all services and stop the server'''
+        if exc_type != KeyboardInterrupt:
+            print(f'Exception was raised:')
+            print(exc_value)
+
+        print('\nSystem was stopped, unregistering services')
+        self.unregister_all_services()
+        print('Stopping server')
+        self.server.stop()
+        print('Shutdown completed')
+
+        return True
+    """
+if __name__ == '__main__':
+    pass

arrowhead_client/configuration.py

@@ -1,14 +1,8 @@
-#!/usr/bin/env python
-
-from typing import Dict
-
 default_config = {
-    'service_registry': {'address': '127.0.0.1', 'port': '8443'},
-    'orchestrator': {'address': '127.0.0.1', 'port': '8441'},
-    'eventhandler': {'address': '127.0.0.1', 'port': '8455'},
-    'gatekeeper': {'address': '127.0.0.1', 'port': '8449'},
-    'gateway': {'address': '127.0.0.1', 'port': '8453'}, }
-
+    'service_registry': {'address': '127.0.0.1', 'port': 8443},
+    'orchestrator': {'address': '127.0.0.1', 'port': 8441},
+    'eventhandler': {'address': '127.0.0.1', 'port': 8455},
+    'gatekeeper': {'address': '127.0.0.1', 'port': 8449},
+    'gateway': {'address': '127.0.0.1', 'port': 8453}, }
 
-def set_config() -> Dict[str, Dict[str, str]]:
-    return default_config
+config = default_config

arrowhead_client/consumer.py

@@ -0,0 +1,44 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from typing import Tuple, Dict, Union
+
+import requests as backend
+from arrowhead_client.service import Service
+from arrowhead_client.system import ArrowheadSystem as System
+
+@dataclass
+class Consumer():
+    """ Class to create Arrowhead consumer systems """
+
+    #TODO: Add all arguments instead of using *args
+    def __init__(self, keyfile, certfile) -> None:
+        self.keyfile = keyfile
+        self.certfile = certfile
+        self._consumed_services: Dict[str, Tuple[Service, System, str]] = {}
+
+    def consume_service(self, service_definition: str, **kwargs) -> backend.Response:
+        """ Consume registered service """
+        # TODO: Add error handling for the case where the service is not
+        # registered in _consumed_services
+
+        uri, http_method = self._service_uri(service_definition)
+
+        service_response = backend.request(http_method, uri, verify=False, **kwargs)
+
+        return service_response
+
+        #TODO: type ignore above should be removed when mypy issue
+        # https://github.com/python/mypy/issues/6799 is fixed
+
+    def _service_uri(self, service_definition: str) -> Tuple[str, str]:
+        service, system, http_method = self._consumed_services[service_definition]
+        uri = f'https://{system.authority}/{service.service_uri}'
+
+        return uri, http_method
+
+    def _extract_payload(self, service_response, interface) -> Union[Dict, str]:
+        if interface.payload.upper() == 'JSON':
+            return service_response.json()
+
+        return service_response.text
+

arrowhead_client/core_service_forms.py

@@ -3,10 +3,9 @@
 
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
-from typing import List, Optional, Dict, Union, Any, Sequence, Mapping
+from typing import List, Optional, Dict, Union, Sequence, Mapping
 
 from . import utils
-from .utils import ServiceInterface
 
 
 class BaseServiceForm(ABC):
@@ -43,8 +42,8 @@ class ServiceQueryForm(CoreSystemServiceForm):
     ping_providers: bool = True
 
     def __post_init__(self):
-        self.interface_requirements = utils.handle_requirements(self.interface_requirements)
-        self.security_requirements = utils.handle_requirements(self.security_requirements)
+        self.interface_requirements = utils.uppercase_strings_in_list(self.interface_requirements)
+        self.security_requirements = utils.uppercase_strings_in_list(self.security_requirements)
 
 
 @dataclass