Add connections

README.md

@@ -63,28 +63,31 @@ of it, and exposes only plain datastructures that reflect the network response.
 ```python
 import httpcore
 
-response = await httpcore.request('GET', 'http://example.com')
+http = httpcore.ConnectionPool()
+response = await http.request('GET', 'http://example.com')
 assert response.status_code == 200
 assert response.body == b'Hello, world'
 ```
 
 Top-level API...
 
 ```python
-response = await httpcore.request(method, url, [headers], [body], [stream])
+http = httpcore.ConnectionPool([ssl], [timeout], [limits])
+response = await http.request(method, url, [headers], [body], [stream])
 ```
 
-Explicit PoolManager...
+ConnectionPool as a context-manager...
 
 ```python
-async with httpcore.PoolManager([ssl], [timeout], [limits]) as pool:
-    response = await pool.request(method, url, [headers], [body], [stream])
+async with httpcore.ConnectionPool([ssl], [timeout], [limits]) as http:
+    response = await http.request(method, url, [headers], [body], [stream])
 ```
 
 Streaming...
 
 ```python
-response = await httpcore.request(method, url, stream=True)
+http = httpcore.ConnectionPool()
+response = await http.request(method, url, stream=True)
 async for part in response.stream():
     ...
 ```
@@ -100,7 +103,7 @@ import httpcore
 class GatewayServer:
     def __init__(self, base_url):
         self.base_url = base_url
-        self.pool = httpcore.PoolManager()
+        self.http = httpcore.ConnectionPool()
 
     async def __call__(self, scope, receive, send):
         assert scope['type'] == 'http'
@@ -122,7 +125,7 @@ class GatewayServer:
                 if not message.get('more_body', False):
                     break
 
-        response = await self.pool.request(
+        response = await self.http.request(
             method, url, headers=headers, body=body, stream=True
         )
 

httpcore/__init__.py

@@ -1,5 +1,6 @@
-from .api import PoolManager, Response, request
 from .config import PoolLimits, SSLConfig, TimeoutConfig
+from .datastructures import URL, Request, Response
 from .exceptions import ResponseClosed, StreamConsumed
+from .pool import ConnectionPool
 
 __version__ = "0.0.2"

httpcore/config.py

@@ -1,5 +1,7 @@
 import typing
 
+import certifi
+
 
 class SSLConfig:
     """
@@ -52,3 +54,4 @@ def __init__(self, *, max_hosts: int, conns_per_host: int, hard_limit: bool):
 DEFAULT_SSL_CONFIG = SSLConfig(cert=None, verify=True)
 DEFAULT_TIMEOUT_CONFIG = TimeoutConfig(timeout=5.0)
 DEFAULT_POOL_LIMITS = PoolLimits(max_hosts=10, conns_per_host=10, hard_limit=False)
+DEFAULT_CA_BUNDLE_PATH = certifi.where()

httpcore/connections.py

@@ -0,0 +1,129 @@
+import asyncio
+import ssl
+import typing
+
+import h11
+
+from .config import TimeoutConfig
+from .datastructures import Request, Response
+from .exceptions import ConnectTimeout, ReadTimeout
+
+H11Event = typing.Union[
+    h11.Request,
+    h11.Response,
+    h11.InformationalResponse,
+    h11.Data,
+    h11.EndOfMessage,
+    h11.ConnectionClosed,
+]
+
+
+class Connection:
+    def __init__(self, timeout: TimeoutConfig):
+        self.reader = None
+        self.writer = None
+        self.state = h11.Connection(our_role=h11.CLIENT)
+        self.timeout = timeout
+
+    async def open(
+        self,
+        hostname: str,
+        port: int,
+        *,
+        ssl: typing.Union[bool, ssl.SSLContext] = False
+    ) -> None:
+        try:
+            self.reader, self.writer = await asyncio.wait_for(  # type: ignore
+                asyncio.open_connection(hostname, port, ssl=ssl),
+                self.timeout.connect_timeout,
+            )
+        except asyncio.TimeoutError:
+            raise ConnectTimeout()
+
+    async def send(self, request: Request, stream: bool = False) -> Response:
+        method = request.method.encode()
+        target = request.url.target
+        host_header = (b"host", request.url.netloc.encode("ascii"))
+        if request.is_streaming:
+            content_length = (b"transfer-encoding", b"chunked")
+        else:
+            content_length = (b"content-length", str(len(request.body)).encode())
+
+        headers = [host_header, content_length] + request.headers
+
+        #  Start sending the request.
+        event = h11.Request(method=method, target=target, headers=headers)
+        await self._send_event(event)
+
+        # Send the request body.
+        if request.is_streaming:
+            async for data in request.stream():
+                event = h11.Data(data=data)
+                await self._send_event(event)
+        elif request.body:
+            event = h11.Data(data=request.body)
+            await self._send_event(event)
+
+        # Finalize sending the request.
+        event = h11.EndOfMessage()
+        await self._send_event(event)
+
+        # Start getting the response.
+        event = await self._receive_event()
+        if isinstance(event, h11.InformationalResponse):
+            event = await self._receive_event()
+        assert isinstance(event, h11.Response)
+        status_code = event.status_code
+        headers = event.headers
+
+        if stream:
+            body_iter = self.body_iter()
+            return Response(status_code=status_code, headers=headers, body=body_iter)
+
+        #  Get the response body.
+        body = b""
+        event = await self._receive_event()
+        while isinstance(event, h11.Data):
+            body += event.data
+            event = await self._receive_event()
+        assert isinstance(event, h11.EndOfMessage)
+        await self.close()
+
+        return Response(status_code=status_code, headers=headers, body=body)
+
+    async def body_iter(self) -> typing.AsyncIterator[bytes]:
+        event = await self._receive_event()
+        while isinstance(event, h11.Data):
+            yield event.data
+            event = await self._receive_event()
+        assert isinstance(event, h11.EndOfMessage)
+        await self.close()
+
+    async def _send_event(self, event: H11Event) -> None:
+        assert self.writer is not None
+
+        data = self.state.send(event)
+        self.writer.write(data)
+
+    async def _receive_event(self) -> H11Event:
+        assert self.reader is not None
+
+        event = self.state.next_event()
+
+        while event is h11.NEED_DATA:
+            try:
+                data = await asyncio.wait_for(
+                    self.reader.read(2048), self.timeout.read_timeout
+                )
+            except asyncio.TimeoutError:
+                raise ReadTimeout()
+            self.state.receive_data(data)
+            event = self.state.next_event()
+
+        return event
+
+    async def close(self) -> None:
+        if self.writer is not None:
+            self.writer.close()
+            if hasattr(self.writer, "wait_closed"):
+                await self.writer.wait_closed()