WIP: using a queue to implement stream

curl_cffi/requests/models.py

@@ -51,6 +51,7 @@ def __init__(self, curl: Optional[Curl] = None, request: Optional[Request] = Non
         self.redirect_url = ""
         self.http_version = 0
         self.history = []
+        self.queue = None
 
     @property
     def text(self) -> str:
@@ -60,6 +61,14 @@ def raise_for_status(self):
         if not self.ok:
             raise RequestsError(f"HTTP Error {self.status_code}: {self.reason}")
 
+    def iter(self):
+        while True:
+            chunk = self.queue.get()  # type: ignore
+            if chunk is None:
+                self.curl.reset()
+                return
+            yield chunk
+
     def json(self, **kw):
         return loads(self.content, **kw)
 

curl_cffi/requests/session.py

@@ -3,12 +3,14 @@
 import re
 import threading
 import warnings
+import queue
 from enum import Enum
 from functools import partialmethod
 from io import BytesIO
 from json import dumps
 from typing import Callable, Dict, List, Optional, Tuple, Union, cast
 from urllib.parse import ParseResult, parse_qsl, unquote, urlencode, urlparse
+from concurrent.futures import ThreadPoolExecutor
 
 from .. import AsyncCurl, Curl, CurlError, CurlInfo, CurlOpt, CurlHttpVersion
 from .cookies import Cookies, CookieTypes, CurlMorsel
@@ -185,6 +187,7 @@ def _set_curl_options(
         default_headers: Optional[bool] = None,
         http_version: Optional[CurlHttpVersion] = None,
         interface: Optional[str] = None,
+        stream: bool = False,
     ):
         c = curl
 
@@ -355,12 +358,14 @@ def _set_curl_options(
         for k, v in self.curl_options.items():
             c.setopt(k, v)
 
-        if content_callback is None:
+        buffer = None
+        if stream:
+            c.setopt(CurlOpt.WRITEFUNCTION, self.queue.put)  # type: ignore
+        elif content_callback is not None:
+            c.setopt(CurlOpt.WRITEFUNCTION, content_callback)
+        else:
             buffer = BytesIO()
             c.setopt(CurlOpt.WRITEDATA, buffer)
-        else:
-            buffer = None
-            c.setopt(CurlOpt.WRITEFUNCTION, content_callback)
         header_buffer = BytesIO()
         c.setopt(CurlOpt.HEADERDATA, header_buffer)
 
@@ -470,6 +475,8 @@ def __init__(
         super().__init__(**kwargs)
         self._thread = thread
         self._use_thread_local_curl = use_thread_local_curl
+        self._queue = None
+        self._executor = None
         if use_thread_local_curl:
             self._local = threading.local()
             if curl:
@@ -492,6 +499,18 @@ def curl(self):
         else:
             return self._curl
 
+    @property
+    def executor(self):
+        if self._executor is None:
+            self._executor = ThreadPoolExecutor()
+        return self._executor
+
+    @property
+    def queue(self):
+        if self._queue is None:
+            self._queue = queue.Queue()
+        return self._queue
+
     def __enter__(self):
         return self
 
@@ -525,6 +544,7 @@ def request(
         default_headers: Optional[bool] = None,
         http_version: Optional[CurlHttpVersion] = None,
         interface: Optional[str] = None,
+        stream: bool = False
     ) -> Response:
         """Send the request, see [curl_cffi.requests.request](/api/curl_cffi.requests/#curl_cffi.requests.request) for details on parameters."""
         c = self.curl
@@ -551,6 +571,7 @@ def request(
             default_headers=default_headers,
             http_version=http_version,
             interface=interface,
+            stream=stream,
         )
         try:
             if self._thread == "eventlet":
@@ -560,17 +581,25 @@ def request(
                 # see: https://www.gevent.org/api/gevent.threadpool.html
                 gevent.get_hub().threadpool.spawn(c.perform).get()
             else:
-                c.perform()
+                if stream:
+                    def perform():
+                        c.perform()
+                        self.queue.put(None)
+                    self.executor.submit(perform)
+                else:
+                    c.perform()
         except CurlError as e:
             rsp = self._parse_response(c, buffer, header_buffer)
             rsp.request = req
             raise RequestsError(str(e), e.code, rsp) from e
         else:
             rsp = self._parse_response(c, buffer, header_buffer)
             rsp.request = req
+            rsp.queue = self.queue
             return rsp
         finally:
-            self.curl.reset()
+            if not stream:
+                self.curl.reset()
 
     head = partialmethod(request, "HEAD")
     get = partialmethod(request, "GET")