ves_app.py 文件源码

def send_data(self, event):
        """Send event to VES"""
        server_url = "http{}://{}:{}{}/eventListener/v{}{}".format(
            's' if self._app_config['UseHttps'] else '',
            self._app_config['Domain'], int(self._app_config['Port']),
            '{}'.format('/{}'.format(self._app_config['Path']) if len(
                self._app_config['Path']) > 0 else ''),
            int(self._app_config['ApiVersion']), '{}'.format(
                '/{}'.format(self._app_config['Topic']) if len(
                    self._app_config['Topic']) > 0 else ''))
        logging.info('Vendor Event Listener is at: {}'.format(server_url))
        credentials = base64.b64encode('{}:{}'.format(
            self._app_config['Username'],
            self._app_config['Password']).encode()).decode()
        logging.info('Authentication credentials are: {}'.format(credentials))
        try:
            request = url.Request(server_url)
            request.add_header('Authorization', 'Basic {}'.format(credentials))
            request.add_header('Content-Type', 'application/json')
            event_str = json.dumps(event).encode()
            logging.debug("Sending {} to {}".format(event_str, server_url))
            url.urlopen(request, event_str, timeout=1)
            logging.debug("Sent data to {} successfully".format(server_url))
        except url.HTTPError as e:
            logging.error('Vendor Event Listener exception: {}'.format(e))
        except url.URLError as e:
            logging.error(
                'Vendor Event Listener is is not reachable: {}'.format(e))
        except Exception as e:
            logging.error('Vendor Event Listener error: {}'.format(e))