trying aiohttp

Author: Ilija Vukotic

pyproject.toml

@@ -9,6 +9,7 @@ python = "^3.6"
 pika = "1.1.0"
 make-it-sync = "^1.0.0"
 requests = "^2.25.0"
+aiohttp = "^3.8.1"
 
 [tool.poetry.dev-dependencies]
 pytest = "^5.2"

src/servicex_did_finder_lib/communication.py

@@ -44,7 +44,7 @@ def add(self, file_info: Dict[str, Any]):
             self._summary.add_file(file_info)
             if self._summary.file_count == 1:
                 self._servicex.post_transform_start()
-            # self._servicex.put_file_add(file_info)
+            self._servicex.put_file_add(file_info)
 
     def send_on(self, count):
         'Send the accumulated files'

src/servicex_did_finder_lib/servicex_adaptor.py

@@ -27,7 +27,10 @@
 # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 import json
 from datetime import datetime
+
 import requests
+import aiohttp
+
 import logging
 
 
@@ -65,7 +68,7 @@ def post_status_update(self, status_msg, severity="info"):
     def _prefix_file(self, file_path):
         return file_path if not self.file_prefix else self.file_prefix+file_path
 
-    def put_file_add(self, file_info):
+    async def put_file_add(self, file_info):
         success = False
         attempts = 0
         while not success and attempts < MAX_RETRIES:
@@ -77,10 +80,15 @@ def put_file_add(self, file_info):
                     'file_size': file_info['file_size'],
                     'file_events': file_info['file_events']
                 }
-                requests.put(self.endpoint + "/files", json=mesg)
+
+                async with aiohttp.ClientSession() as session:
+                    async with session.put(self.endpoint + "/files", json=mesg) as response:
+                        if response.status != 200:
+                            self.logger.error(
+                                'could not send a put_file {}'.format(response.status))
                 self.logger.info(f"Metric: {json.dumps(mesg)}")
                 success = True
-            except requests.exceptions.ConnectionError:
+            except aiohttp.ClientConnectorError:
                 self.logger.exception(f'Connection error to ServiceX App. Will retry '
                                       f'(try {attempts} out of {MAX_RETRIES}')
                 attempts += 1