1- from typing import Any , List , Union
1+ from typing import Any , List , Union , Optional
2+ import asyncio
23
34import aiohttp
45
56
7+ # Global session and lock for thread-safe initialization
8+ _client_session : Optional [aiohttp .ClientSession ] = None
9+ _session_lock : asyncio .Lock = asyncio .Lock ()
10+
11+
12+ async def _get_session () -> aiohttp .ClientSession :
13+ """Get or create the shared ClientSession.
14+
15+ Returns
16+ -------
17+ aiohttp.ClientSession
18+ The shared client session with configured timeout and connection pooling.
19+ """
20+ global _client_session
21+
22+ # Double-check locking pattern for async
23+ if _client_session is None or _client_session .closed :
24+ async with _session_lock :
25+ # Check again after acquiring lock
26+ if _client_session is None or _client_session .closed :
27+ # Configure timeout optimized for localhost IPC
28+ timeout = aiohttp .ClientTimeout (
29+ total = 240 , # 4-minute total timeout for slow operations
30+ sock_connect = 10 , # Fast connection over localhost
31+ sock_read = None # Covered by total timeout
32+ )
33+
34+ # Configure TCP connector optimized for localhost IPC
35+ connector = aiohttp .TCPConnector (
36+ limit = 30 , # Maximum connections for single host
37+ limit_per_host = 30 , # Maximum connections per host
38+ enable_cleanup_closed = True # Enable cleanup of closed connections
39+ )
40+
41+ _client_session = aiohttp .ClientSession (
42+ timeout = timeout ,
43+ connector = connector
44+ )
45+
46+ return _client_session
47+
48+
49+ async def _handle_request_error ():
50+ """Handle connection errors by closing and resetting the session.
51+
52+ This handles cases where the remote host process recycles.
53+ """
54+ global _client_session
55+ async with _session_lock :
56+ if _client_session is not None and not _client_session .closed :
57+ try :
58+ await _client_session .close ()
59+ finally :
60+ _client_session = None
61+
62+
63+ async def _close_session () -> None :
64+ """Close the shared ClientSession if it exists.
65+
66+ Note: This function is currently only called by _handle_request_error().
67+ There is no worker shutdown hook available, but process shutdown will
68+ clean up all resources automatically.
69+ """
70+ global _client_session
71+
72+ async with _session_lock :
73+ if _client_session is not None and not _client_session .closed :
74+ try :
75+ await _client_session .close ()
76+ finally :
77+ _client_session = None
78+
79+
680async def post_async_request (url : str ,
781 data : Any = None ,
882 trace_parent : str = None ,
9- trace_state : str = None ) -> List [Union [int , Any ]]:
83+ trace_state : str = None ,
84+ function_invocation_id : str = None ) -> List [Union [int , Any ]]:
1085 """Post request with the data provided to the url provided.
1186
1287 Parameters
@@ -19,62 +94,96 @@ async def post_async_request(url: str,
1994 traceparent header to send with the request
2095 trace_state: str
2196 tracestate header to send with the request
97+ function_invocation_id: str
98+ function invocation ID header to send for correlation
2299
23100 Returns
24101 -------
25102 [int, Any]
26103 Tuple with the Response status code and the data returned from the request
27104 """
28- async with aiohttp .ClientSession () as session :
29- headers = {}
30- if trace_parent :
31- headers ["traceparent" ] = trace_parent
32- if trace_state :
33- headers ["tracestate" ] = trace_state
105+ session = await _get_session ()
106+ headers = {}
107+ if trace_parent :
108+ headers ["traceparent" ] = trace_parent
109+ if trace_state :
110+ headers ["tracestate" ] = trace_state
111+ if function_invocation_id :
112+ headers ["X-Azure-Functions-InvocationId" ] = function_invocation_id
113+
114+ try :
34115 async with session .post (url , json = data , headers = headers ) as response :
35116 # We disable aiohttp's input type validation
36117 # as the server may respond with alternative
37118 # data encodings. This is potentially unsafe.
38119 # More here: https://docs.aiohttp.org/en/stable/client_advanced.html
39120 data = await response .json (content_type = None )
40121 return [response .status , data ]
122+ except (aiohttp .ClientError , asyncio .TimeoutError ):
123+ # On connection errors, close and recreate session for next request
124+ await _handle_request_error ()
125+ raise
41126
42127
43- async def get_async_request (url : str ) -> List [Any ]:
128+ async def get_async_request (url : str ,
129+ function_invocation_id : str = None ) -> List [Any ]:
44130 """Get the data from the url provided.
45131
46132 Parameters
47133 ----------
48134 url: str
49135 url to get the data from
136+ function_invocation_id: str
137+ function invocation ID header to send for correlation
50138
51139 Returns
52140 -------
53141 [int, Any]
54142 Tuple with the Response status code and the data returned from the request
55143 """
56- async with aiohttp .ClientSession () as session :
57- async with session .get (url ) as response :
144+ session = await _get_session ()
145+ headers = {}
146+ if function_invocation_id :
147+ headers ["X-Azure-Functions-InvocationId" ] = function_invocation_id
148+
149+ try :
150+ async with session .get (url , headers = headers ) as response :
58151 data = await response .json (content_type = None )
59152 if data is None :
60153 data = ""
61154 return [response .status , data ]
155+ except (aiohttp .ClientError , asyncio .TimeoutError ):
156+ # On connection errors, close and recreate session for next request
157+ await _handle_request_error ()
158+ raise
62159
63160
64- async def delete_async_request (url : str ) -> List [Union [int , Any ]]:
161+ async def delete_async_request (url : str ,
162+ function_invocation_id : str = None ) -> List [Union [int , Any ]]:
65163 """Delete the data from the url provided.
66164
67165 Parameters
68166 ----------
69167 url: str
70168 url to delete the data from
169+ function_invocation_id: str
170+ function invocation ID header to send for correlation
71171
72172 Returns
73173 -------
74174 [int, Any]
75175 Tuple with the Response status code and the data returned from the request
76176 """
77- async with aiohttp .ClientSession () as session :
78- async with session .delete (url ) as response :
177+ session = await _get_session ()
178+ headers = {}
179+ if function_invocation_id :
180+ headers ["X-Azure-Functions-InvocationId" ] = function_invocation_id
181+
182+ try :
183+ async with session .delete (url , headers = headers ) as response :
79184 data = await response .json (content_type = None )
80185 return [response .status , data ]
186+ except (aiohttp .ClientError , asyncio .TimeoutError ):
187+ # On connection errors, close and recreate session for next request
188+ await _handle_request_error ()
189+ raise
0 commit comments