bind out write data api

Author: azkrishpy

awscrt/http.py

@@ -281,7 +281,8 @@ def new(cls,
     def request(self,
                 request: 'HttpRequest',
                 on_response: Optional[Callable[..., None]] = None,
-                on_body: Optional[Callable[..., None]] = None) -> 'HttpClientStream':
+                on_body: Optional[Callable[..., None]] = None,
+                manual_write: bool = False) -> 'HttpClientStream':
         """Create :class:`HttpClientStream` to carry out the request/response exchange.
 
         NOTE: The HTTP stream sends no data until :meth:`HttpClientStream.activate()`
@@ -320,10 +321,14 @@ def request(self,
                 An exception raise by this function will cause the HTTP stream to end in error.
                 This callback is always invoked on the connection's event-loop thread.
 
+            manual_write (bool): If True, enables manual data writing on the stream.
+                This allows calling :meth:`HttpClientStream.write_data()` to stream
+                the request body in chunks. Works for both HTTP/1.1 and HTTP/2.
+
         Returns:
             HttpClientStream:
         """
-        return HttpClientStream(self, request, on_response, on_body)
+        return HttpClientStream(self, request, on_response, on_body, manual_write)
 
     def close(self) -> "concurrent.futures.Future":
         """Close the connection.
@@ -586,6 +591,40 @@ def update_window(self, increment_size: int) -> None:
         """
         _awscrt.http_stream_update_window(self, increment_size)
 
+    def write_data(self,
+                   data_stream: Union[InputStream, Any],
+                   end_stream: bool = False) -> "concurrent.futures.Future":
+        '''Write a chunk of data to the request body stream.
+
+        Works for both HTTP/1.1 and HTTP/2 streams.
+        The stream must have been created with ``manual_write=True``.
+        You must call :meth:`activate()` before using this method.
+
+        Args:
+            data_stream (InputStream): Data to write. If not an InputStream,
+                it will be wrapped in one. Can be None to write zero bytes.
+
+            end_stream (bool): True to indicate this is the last chunk and no more data
+                will be sent. False if more chunks will follow.
+
+        Returns:
+            concurrent.futures.Future: Future that completes when the write operation
+                is done. The future will contain None on success, or an exception on failure.
+        '''
+        future = Future()
+        body_stream = InputStream.wrap(data_stream, allow_none=True)
+
+        def on_write_complete(error_code: int) -> None:
+            if future.cancelled():
+                return
+            if error_code:
+                future.set_exception(awscrt.exceptions.from_code(error_code))
+            else:
+                future.set_result(None)
+
+        _awscrt.http_stream_write_data(self, body_stream, end_stream, on_write_complete)
+        return future
+
 
 class HttpClientStream(HttpClientStreamBase):
     """HTTP stream that sends a request and receives a response.
@@ -608,8 +647,9 @@ def __init__(self,
                  connection: HttpClientConnection,
                  request: 'HttpRequest',
                  on_response: Optional[Callable[..., None]] = None,
-                 on_body: Optional[Callable[..., None]] = None) -> None:
-        self._init_common(connection, request, on_response, on_body)
+                 on_body: Optional[Callable[..., None]] = None,
+                 manual_write: bool = False) -> None:
+        self._init_common(connection, request, on_response, on_body, manual_write)
 
     def activate(self) -> None:
         """Begin sending the request.
@@ -683,16 +723,16 @@ def activate(self) -> None:
     def write_data(self,
                    data_stream: Union[InputStream, Any],
                    end_stream: bool = False) -> "concurrent.futures.Future":
-        """Write a chunk of data to the request body stream.
+        '''Write a chunk of data to the request body stream.
 
         This method is only available when the stream was created with
         manual_write=True. This allows incremental writing of request data.
 
-        Note: In the asyncio version, this is replaced by the request_body_generator parameter
-        which accepts an async generator.
+        Note: This method is inherited from the base class and works for both
+        HTTP/1.1 and HTTP/2. It is kept here for backward compatibility.
 
         Args:
-            data_stream (Union[InputStream, Any]): Data to write. If not an InputStream,
+            data_stream (InputStream): Data to write. If not an InputStream,
                 it will be wrapped in one. Can be None to send an empty chunk.
 
             end_stream (bool): True to indicate this is the last chunk and no more data
@@ -701,21 +741,8 @@ def write_data(self,
         Returns:
             concurrent.futures.Future: Future that completes when the write operation
                 is done. The future will contain None on success, or an exception on failure.
-        """
-        future = Future()
-        body_stream = InputStream.wrap(data_stream, allow_none=True)
-
-        def on_write_complete(error_code: int) -> None:
-            if future.cancelled():
-                # the future was cancelled, so we don't need to set the result or exception
-                return
-            if error_code:
-                future.set_exception(awscrt.exceptions.from_code(error_code))
-            else:
-                future.set_result(None)
-
-        _awscrt.http2_client_stream_write_data(self, body_stream, end_stream, on_write_complete)
-        return future
+        '''
+        return super().write_data(data_stream, end_stream)
 
 
 class HttpMessageBase(NativeResource):

crt/aws-c-http

@@ -50,6 +50,7 @@ PyObject *aws_py_http_client_stream_new(PyObject *self, PyObject *args);
 PyObject *aws_py_http_client_stream_activate(PyObject *self, PyObject *args);
 
 PyObject *aws_py_http2_client_stream_write_data(PyObject *self, PyObject *args);
+PyObject *aws_py_http_stream_write_data(PyObject *self, PyObject *args);
 
 /* Create capsule around new request-style aws_http_message struct */
 PyObject *aws_py_http_message_new_request(PyObject *self, PyObject *args);

source/http.h

@@ -303,6 +303,7 @@ PyObject *aws_py_http_client_stream_new(PyObject *self, PyObject *args) {
         .on_complete = s_on_stream_complete,
         .on_h2_remote_end_stream = s_on_h2_remote_end_stream,
         .user_data = stream,
+        .use_manual_data_writes = http2_manual_write,
         .http2_use_manual_data_writes = http2_manual_write,
     };
 
@@ -410,3 +411,63 @@ PyObject *aws_py_http2_client_stream_write_data(PyObject *self, PyObject *args)
     }
     Py_RETURN_NONE;
 }
+
+static void s_on_http_stream_write_data_complete(struct aws_http_stream *stream, int error_code, void *user_data) {
+    (void)stream;
+    PyObject *py_on_write_complete = (PyObject *)user_data;
+    AWS_FATAL_ASSERT(py_on_write_complete);
+    PyGILState_STATE state;
+    if (aws_py_gilstate_ensure(&state)) {
+        return; /* Python has shut down. Nothing matters anymore, but don't crash */
+    }
+
+    PyObject *result = PyObject_CallFunction(py_on_write_complete, "(i)", error_code);
+    if (result) {
+        Py_DECREF(result);
+    } else {
+        PyErr_WriteUnraisable(PyErr_Occurred());
+    }
+    Py_DECREF(py_on_write_complete);
+    PyGILState_Release(state);
+}
+
+PyObject *aws_py_http_stream_write_data(PyObject *self, PyObject *args) {
+    (void)self;
+
+    PyObject *py_stream = NULL;
+    PyObject *py_body_stream = NULL;
+    int end_stream = false;
+    PyObject *py_on_write_complete = NULL;
+    if (!PyArg_ParseTuple(args, "OOpO", &py_stream, &py_body_stream, &end_stream, &py_on_write_complete)) {
+        return NULL;
+    }
+
+    struct aws_http_stream *http_stream = aws_py_get_http_stream(py_stream);
+    if (!http_stream) {
+        return NULL;
+    }
+
+    struct aws_input_stream *body_stream = NULL;
+    if (py_body_stream != Py_None) {
+        body_stream = aws_py_get_input_stream(py_body_stream);
+        if (!body_stream) {
+            return PyErr_AwsLastError();
+        }
+    }
+
+    Py_INCREF(py_on_write_complete);
+
+    struct aws_http_stream_write_data_options write_options = {
+        .data = body_stream,
+        .end_stream = end_stream,
+        .on_complete = s_on_http_stream_write_data_complete,
+        .user_data = py_on_write_complete,
+    };
+
+    int error = aws_http_stream_write_data(http_stream, &write_options);
+    if (error) {
+        Py_DECREF(py_on_write_complete);
+        return PyErr_AwsLastError();
+    }
+    Py_RETURN_NONE;
+}

source/http_stream.c

@@ -1052,6 +1052,7 @@ static PyMethodDef s_module_methods[] = {
     AWS_PY_METHOD_DEF(http_client_stream_new, METH_VARARGS),
     AWS_PY_METHOD_DEF(http_client_stream_activate, METH_VARARGS),
     AWS_PY_METHOD_DEF(http2_client_stream_write_data, METH_VARARGS),
+    AWS_PY_METHOD_DEF(http_stream_write_data, METH_VARARGS),
     AWS_PY_METHOD_DEF(http_message_new_request, METH_VARARGS),
     AWS_PY_METHOD_DEF(http_message_get_request_method, METH_VARARGS),
     AWS_PY_METHOD_DEF(http_message_set_request_method, METH_VARARGS),