feat: custom session handler

examples/custom_ssl_session.py

@@ -0,0 +1,30 @@
+from reolinkapi import Camera
+
+import urllib3
+import requests
+from urllib3.util import create_urllib3_context
+
+class CustomSSLContextHTTPAdapter(requests.adapters.HTTPAdapter):
+    def __init__(self, ssl_context=None, **kwargs):
+        self.ssl_context = ssl_context
+        super().__init__(**kwargs)
+
+    def init_poolmanager(self, connections, maxsize, block=False):
+        self.poolmanager = urllib3.poolmanager.PoolManager(
+            num_pools=connections, maxsize=maxsize,
+            block=block, ssl_context=self.ssl_context)
+
+urllib3.disable_warnings()
+ctx = create_urllib3_context()
+ctx.load_default_certs()
+ctx.set_ciphers("AES128-GCM-SHA256")
+ctx.check_hostname = False
+
+session = requests.session()
+session.adapters.pop("https://", None)
+session.mount("https://", CustomSSLContextHTTPAdapter(ctx))
+
+## Add a custom http handler to add in different ciphers that may
+## not be aloud by default in openssl which urlib uses
+cam = Camera("url", "user", "password", https=True, session=session)
+cam.reboot_camera()

reolinkapi/handlers/api_handler.py

@@ -58,6 +58,7 @@ def __init__(self, ip: str, username: str, password: str, https: bool = False, *
         self.username = username
         self.password = password
         Request.proxies = kwargs.get("proxy")  # Defaults to None if key isn't found
+        Request.session = kwargs.get("session")  # Defaults to None if key isn't found
 
     def login(self) -> bool:
         """

reolinkapi/handlers/rest_handler.py

@@ -4,6 +4,14 @@
 
 class Request:
     proxies = None
+    session = None
+
+    @staticmethod
+    def __getSession():
+        reqHandler = requests
+        if Request.session is not None:
+            reqHandler = Request.session
+        return reqHandler
 
     @staticmethod
     def post(url: str, data: List[Dict], params: Dict[str, Union[str, float]] = None) -> \
@@ -17,7 +25,7 @@ def post(url: str, data: List[Dict], params: Dict[str, Union[str, float]] = None
         """
         try:
             headers = {'content-type': 'application/json'}
-            r = requests.post(url, verify=False, params=params, json=data, headers=headers,
+            r = Request.__getSession().post(url, verify=False, params=params, json=data, headers=headers,
                               proxies=Request.proxies)
             if r.status_code == 200:
                 return r
@@ -37,7 +45,7 @@ def get(url: str, params: Dict[str, Union[str, float]], timeout: float = 1) -> O
         :return:
         """
         try:
-            data = requests.get(url=url, verify=False, params=params, timeout=timeout, proxies=Request.proxies)
+            data = Request.__getSession().get(url=url, verify=False, params=params, timeout=timeout, proxies=Request.proxies)
             return data
         except Exception as e:
             print("Get Error\n", e)