improve/http2

.github/workflows/python.yml

@@ -77,6 +77,10 @@ jobs:
         run: npm install
         working-directory: tests
 
+      - name: Lint with ruff
+        run: |
+          poetry run ruff check aiosonic tests aiosonic_utils
+
       - name: Test with pytest
         run: |
           poetry run py.test --cov-append

AGENTS.md

@@ -2,13 +2,13 @@
 
 ## Build/Lint/Test Commands
 
-ALWAYS USE poetry for commands eg: `poetry run <command>` it coud be "python", py.test, black, etc.
+ALWAYS USE poetry for commands eg: `poetry run <command>` it could be "python", py.test, ruff, etc.
 
 - **Build**: `poetry build` or `make build`
 - **Test all**: `poetry run py.test`
 - **Test single file**: `poetry run py.test tests/test_filename.py`
 - **Test single function**: `poetry run py.test tests/test_filename.py::test_function_name`
-- **Lint/Format**: `poetry run black .` (formatting), `poetry run ruff check .` (linting)
+- **Lint/Format**: `poetry run ruff format .` (formatting), `poetry run ruff check .` (linting)
 - **CI test command**: `poetry run py.test --cov-append`
 - **Run scripts or examples**: `poetry run <file.py>`
 
@@ -19,7 +19,7 @@ ALWAYS USE poetry for commands eg: `poetry run <command>` it coud be "python", p
 ### Error Handling: Custom exceptions in exceptions.py, descriptive messages, appropriate try/except
 ### Documentation: Google/NumPy docstring format for all classes and public methods
 ### Comments: DO NOT ADD ***ANY*** COMMENTS in "aiosonic" package functions or classes
-### Formatting: Black formatter (88 char lines, 4 space indent, no trailing whitespace)
+### Formatting: Ruff formatter (119 char lines, 4 space indent, no trailing whitespace)
 
 ## HTTP Client Architecture
 

aiosonic/__init__.py

@@ -1,4 +1,4 @@
-from aiosonic.base_client import AioSonicBaseClient, BaseClient
-from aiosonic.client import *
-from aiosonic.sse_client import *
-from aiosonic.web_socket_client import *
+from aiosonic.base_client import AioSonicBaseClient as AioSonicBaseClient, BaseClient as BaseClient
+from aiosonic.client import *  # noqa: F403
+from aiosonic.sse_client import *  # noqa: F403
+from aiosonic.web_socket_client import *  # noqa: F403

aiosonic/base_client.py

@@ -66,8 +66,7 @@ class AioSonicBaseClient(BaseClient):
 
     def __init__(self, http_client: Optional[HTTPClient] = None):
         warnings.warn(
-            "AioSonicBaseClient is deprecated and will be removed in 1.x.x releases. "
-            "Use BaseClient instead.",
+            "AioSonicBaseClient is deprecated and will be removed in 1.x.x releases. Use BaseClient instead.",
             DeprecationWarning,
             stacklevel=2,
         )

aiosonic/client.py

@@ -48,9 +48,7 @@
 from aiosonic_utils.structures import CaseInsensitiveDict
 
 # VARIABLES
-_HTTP_RESPONSE_STATUS_LINE = re.compile(
-    r"HTTP/(?P<version>(\d.)?(\d)) (?P<code>\d+) (?P<reason>[\w]*)"
-)
+_HTTP_RESPONSE_STATUS_LINE = re.compile(r"HTTP/(?P<version>(\d.)?(\d)) (?P<code>\d+) (?P<reason>[\w]*)")
 _CHARSET_RGX = re.compile(r"charset=(?P<charset>[\w-]*);?")
 _CHUNK_SIZE = 1024 * 4  # 4kilobytes
 CRLF = "\r\n"
@@ -102,6 +100,9 @@ def __init__(self):
         self.compressed = b""
         self.chunks_readed = False
         self.request_meta = {}
+        self._h2_chunk_queue = None
+        self._h2_sem_release = None
+        self._h2_flow_cb = None
 
     def _set_response_initial(self, data: bytes):
         """Parse first bytes from http response."""
@@ -132,9 +133,7 @@ def logparse(data):
             to_log_info = [[key, val] for key, val in info]
             meta_log = logparse(to_log_info)
             headers_log = logparse(self.raw_headers)
-            dlogger.debug(
-                meta_log + CRLF + "Headers:" + CRLF * 2 + headers_log + "---"
-            )  # noqa
+            dlogger.debug(meta_log + CRLF + "Headers:" + CRLF * 2 + headers_log + "---")  # noqa
 
     def _update_cookies(self, header_tuple):
         """Update jar of cookies."""
@@ -145,11 +144,23 @@ def _set_connection(self, connection: Connection):
         """Set header to response."""
         self._connection = connection
 
+    def _set_h2_queue(self, queue, sem_release, flow_cb):
+        """Attach an HTTP/2 per-stream chunk queue to this response."""
+        self._h2_chunk_queue = queue
+        self._h2_sem_release = sem_release
+        self._h2_flow_cb = flow_cb
+        self.chunked = True
+
     @property
     def status_code(self) -> int:
         """Get status code."""
         return int(self.response_initial["code"])
 
+    @property
+    def http_version(self) -> str:
+        """Get the negotiated HTTP version string (e.g. '2', '1.1', '1.0')."""
+        return self.response_initial.get("version", "")
+
     @property
     def ok(self) -> bool:
         """Returns True if :attr:`status_code` is 2xx range, False if not."""
@@ -211,6 +222,22 @@ async def json(self, json_decoder=loads) -> dict:
 
     async def read_chunks(self) -> AsyncIterator[bytes]:
         """Read chunks from chunked response."""
+        if self._h2_chunk_queue is not None:
+            try:
+                while not self.chunks_readed:
+                    chunk = await self._h2_chunk_queue.get()
+                    if chunk is None:
+                        break
+                    if self._h2_flow_cb:
+                        self._h2_flow_cb(len(chunk))
+                    yield chunk
+                self.chunks_readed = True
+            finally:
+                if self._h2_sem_release:
+                    self._h2_sem_release()
+                    self._h2_sem_release = None
+            return
+
         if not self._connection:
             raise ConnectionError("missing connection, possible already read response.")
         try:
@@ -230,7 +257,15 @@ async def read_chunks(self) -> AsyncIterator[bytes]:
                 self._connection = None
 
     def __del__(self):
-        # clean it
+        if self._h2_chunk_queue is not None:
+            queue = self._h2_chunk_queue
+            self._h2_chunk_queue = None
+            while not queue.empty():
+                queue.get_nowait()
+            if self._h2_sem_release:
+                self._h2_sem_release()
+                self._h2_sem_release = None
+            return
         if self._connection and self._connection.blocked:
             response_read = self.body
             self._connection.ensure_released(response_read)
@@ -394,9 +429,7 @@ async def _send_multipart(
             total_size += len(disp.encode()) + len(val.encode()) + len(CRLF.encode())
             continue
 
-        to_write = (
-            f'Content-Disposition: form-data; name="{key}"; filename="{filename}"{CRLF}'
-        )
+        to_write = f'Content-Disposition: form-data; name="{key}"; filename="{filename}"{CRLF}'
         if content_type:
             to_write += f"Content-Type: {content_type}{CRLF}"
         to_write += CRLF
@@ -423,9 +456,7 @@ async def _gen():
                 filename = val.filename
                 content_type = val.content_type
             else:
-                yield (
-                    f'Content-Disposition: form-data; name="{key}"{CRLF}{CRLF}'
-                ).encode()
+                yield (f'Content-Disposition: form-data; name="{key}"{CRLF}{CRLF}').encode()
                 yield val.encode() + CRLF.encode()
                 continue
 
@@ -473,9 +504,7 @@ async def _do_request(
     args = url_connect, verify, connect_ssl, timeouts, http2
     async with await connector.acquire(*args) as connection:
         if proxy and urlparsed.scheme == "https" and not connection.proxy_connected:
-            await _proxy_connect(
-                connection, proxy, urlparsed, ssl or get_default_ssl_context()
-            )
+            await _proxy_connect(connection, proxy, urlparsed, ssl or get_default_ssl_context())
 
         to_send = headers_data(connection=connection)
 
@@ -516,9 +545,7 @@ async def _do_request(
             raise ReadTimeout()
 
         # reading headers
-        await response._set_response_headers(
-            http_parser.parse_headers_iterator(connection)
-        )
+        await response._set_response_headers(http_parser.parse_headers_iterator(connection))
 
         size = response.headers.get("content-length")
         chunked = response.headers.get("transfer-encoding", "") == "chunked"
@@ -551,6 +578,8 @@ class HTTPClient:
         * **handle_cookies**: Flag to indicate if keep response cookies in
             client and send them in next requests.
         * **verify_ssl**: Flag to indicate if verify ssl certificates.
+        * **http2**: Flag to enable HTTP/2 for all requests made by this client.
+            Per-request ``http2=True`` also works and takes precedence.
     """
 
     def __init__(
@@ -560,14 +589,16 @@ def __init__(
         verify_ssl: bool = True,
         proxy: Optional[Proxy] = None,
         max_redirects: int = 5,
+        http2: bool = False,
     ):
         """Initialize client options."""
-        self.connector = connector or TCPConnector()
+        self.connector = connector or TCPConnector(http2=http2)
         self.handle_cookies = handle_cookies
         self.cookies_map: Dict[str, cookies.SimpleCookie] = {}
         self.verify_ssl = verify_ssl
         self.proxy = proxy
         self.max_redirects = max_redirects
+        self.http2 = http2
 
     async def __aenter__(self):
         return self
@@ -797,11 +828,10 @@ async def request(
         elif data:
             body = http_parser.setup_body_request(data, headers)
 
-        max_redirects = (
-            max_redirects if max_redirects is not None else self.max_redirects
-        )
+        max_redirects = max_redirects if max_redirects is not None else self.max_redirects
         # if class or request method has false, it will be false
         verify_ssl = verify and self.verify_ssl
+        http2 = http2 or self.http2
         reconnect_times = 3
         while reconnect_times > 0:
             headers_data = partial(
@@ -835,16 +865,14 @@ async def request(
                     self._save_new_cookies(str(urlparsed.hostname), response)
 
                 if follow and response.status_code in {301, 302, 303, 307, 308}:
-                    (urlparsed, method, body, transfer_chunked, max_redirects) = (
-                        self._handle_redirect(
-                            current_urlparsed=urlparsed,
-                            headers=headers,
-                            response=response,
-                            max_redirects=max_redirects,
-                            method=method,
-                            body=body,
-                            transfer_chunked=transfer_chunked,
-                        )
+                    (urlparsed, method, body, transfer_chunked, max_redirects) = self._handle_redirect(
+                        current_urlparsed=urlparsed,
+                        headers=headers,
+                        response=response,
+                        max_redirects=max_redirects,
+                        method=method,
+                        body=body,
+                        transfer_chunked=transfer_chunked,
                     )
                     # continue loop to re-issue the request with updated params
                 else:
@@ -900,9 +928,7 @@ def _handle_redirect(
         status = response.status_code
         original_method = method.upper()
 
-        switch_to_get = status == 303 or (
-            status in {301, 302} and original_method == "POST"
-        )
+        switch_to_get = status == 303 or (status in {301, 302} and original_method == "POST")
 
         if switch_to_get:
             method = "GET"
@@ -946,9 +972,7 @@ async def wait_requests(self, timeout: int = 30):
     def _add_cookies_to_request(self, host: str, headers: HeadersType):
         """Add cookies to request."""
         host_cookies = self.cookies_map.get(host)
-        if host_cookies and not any(
-            [header.lower() == "cookie" for header, _ in headers]
-        ):
+        if host_cookies and not any([header.lower() == "cookie" for header, _ in headers]):
             cookies_str = host_cookies.output(header="Cookie:")
             for cookie_data in cookies_str.split("\r\n"):
                 http_parser.add_header(headers, *cookie_data.split(": ", 1))
@@ -959,9 +983,7 @@ def _save_new_cookies(self, host: str, response: HttpResponse):
             self.cookies_map[host] = response.cookies
 
 
-async def _proxy_connect(
-    connection: Connection, proxy: Proxy, desturl: ParseResult, ssl_context: SSLContext
-):
+async def _proxy_connect(connection: Connection, proxy: Proxy, desturl: ParseResult, ssl_context: SSLContext):
     """Send CONNECT and upgrade connection."""
 
     port = desturl.port or (443 if desturl.scheme == "https" else 80)
@@ -980,9 +1002,7 @@ async def _proxy_connect(
     connect_response = await connection.read(4096)
     if b"200 Connection established" not in connect_response:
         connection.close()
-        raise ConnectionError(
-            f"Failed to establish connection through proxy: {connect_response}"
-        )
+        raise ConnectionError(f"Failed to establish connection through proxy: {connect_response}")
 
     if sys.version_info >= (3, 11):
         await connection.upgrade(ssl_context)
@@ -996,9 +1016,7 @@ async def _proxy_connect(
 async def _update_transport(connection: Connection, ssl_context):
     transport = connection.writer.transport
     protocol = transport.get_protocol()
-    new_transport = await get_loop().start_tls(
-        transport, protocol, ssl_context, server_side=False
-    )
+    new_transport = await get_loop().start_tls(transport, protocol, ssl_context, server_side=False)
 
     writer = connection.writer
     reader = connection.reader