import logging
import time
from netdef.Interfaces.ConcurrentWebRequestInterface import (
ConcurrentWebRequestInterface,
)
from netdef.Sources import BaseSource, Sources
[docs]def setup(shared):
ConcurrentWebRequestSource.DEFAULT_CLIENT_SESSION_TIMEOUT = shared.config.config(
"ConcurrentWebRequestSource",
"client_session_timeout",
ConcurrentWebRequestSource.DEFAULT_CLIENT_SESSION_TIMEOUT,
)
[docs]@Sources.register("ConcurrentWebRequestSource")
class ConcurrentWebRequestSource(BaseSource.BaseSource):
DEFAULT_CLIENT_SESSION_TIMEOUT = 2
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.interface = ConcurrentWebRequestInterface
self._session = None
self.logger = logging.getLogger("ConcurrentWebRequestSource")
self.logger.debug("init %s", self.key)
self.basic_auth_user = None
self.basic_auth_pass = None
self.url_scheme = "http://"
self.base_url = None
self.parse_url(self.key)
[docs] def parse_url(self, url):
if "://" in url:
self.url_scheme, url = url.split("://", 1)
self.url_scheme += "://"
if "@" in url:
login, url = url.split("@", 1)
if ":" in login:
self.basic_auth_user, self.basic_auth_pass = login.split(":", 1)
self.base_url = url
[docs] def get_client_session(self):
return self._session
[docs] def add_client_session(self, session):
self._session = session
[docs] def get_client_session_timeout(self):
return self.DEFAULT_CLIENT_SESSION_TIMEOUT
[docs] def get_commands_list(self):
return []
[docs] def has_basic_auth(self):
return not self.basic_auth_user is None
[docs] def get_basic_auth(self):
return self.basic_auth_user, self.basic_auth_pass
[docs] def build_url(self, url):
return "{}{}{}".format(self.url_scheme, self.base_url, url)
[docs] def get_start_url(self):
return self.build_url("/")
[docs] def has_connect_request(self):
return False
[docs] def get_connect_request(self):
raise NotImplementedError
[docs] def has_poll_request(self):
return False
[docs] def get_poll_request_interval(self):
raise NotImplementedError
[docs] def get_poll_request(self):
raise NotImplementedError
[docs]class Request:
__slots__ = ["method", "url", "params", "data"]
def __init__(self, method, url, params=None, data=None):
self.method = method
self.url = url
self.params = params
self.data = data
def __repr__(self):
return " ".join((self.method, self.url))
[docs]class Result:
__slots__ = ["result"]
def __init__(self, result):
self.result = result