Repository: encode/httpx

Issue:
Use the existing runloop for sync calls in async handler
Hi, this library looks awesome and I really want to use it; but I seem to have hit a blocker and can't seem to find a solution in the docs.

TL;DR: Is there any way to provide our own runloop when using the `sync` calls as looking through the source the sync calls are actually using the async request call.

Basically I'm writing a client api that provides 2 interfaces; an async and sync interface.

The issue that i am having is that if I try to use the sync methods with an existing runloop (provided by Sanic in this scenario), it complains with the exception: `Cannot run the event loop while another loop is running`.

Essentially I'm just calling:
```
import httpx
...
class SomeClass(object):
    def get_some_url(self):
        r = httpx.get('some url')
```

note that nothing here is async; apart from the fact that the `.get_some_url` call is made from within an async function like so:

```
import SomeClass

async def handler(request):
    o = SomeClass()
    o.get_some_url()
```

---

This works fine when using the async calls like this:

```
import httpx
...
class SomeClass(object):
    async def get_some_url(self):
        client = httpx.AsyncClient()
        r = await client.get('some url')
```

```
import SomeClass

async def handler(request):
    o = SomeClass()
    await o.get_some_url()
```

So I was just wondering if its possible to provide our own runloop when using the sync calls? Sanic uses uvloop internally, would be nice to be able to just pass that straight in somehow when constructing the `httpx` object, or setting it as a property after import.

Thanks!

Discussion:

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 34d47ec..3734b1f 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,12 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
+## 0.7.8 (November 17, 2019)
+
+### Added
+
+- Add support for proxy tunnels for Python 3.6 + asyncio. (Pull #521)
+
 ## 0.7.7 (November 15, 2019)
 
 ### Fixed
diff --git a/docs/advanced.md b/docs/advanced.md
index 3e73f5b..1365645 100644
--- a/docs/advanced.md
+++ b/docs/advanced.md
@@ -179,6 +179,21 @@ password example-password
 ...
 ```
 
+## Unix Domain Sockets
+
+You can configure an `httpx` client to connect through a unix domain socket via the `uds` parameter. This is useful when making requests to a server that is bound to a socket file rather than an IP address.
+
+Here's an example requesting the Docker Engine API:
+
+```python
+import httpx
+
+
+with httpx.Client(uds="/var/run/docker.sock") as client:
+    # This request will connect through the socket file.
+    resp = client.get("http://localhost/version")
+```
+
 ## HTTP Proxying
 
 HTTPX supports setting up proxies the same way that Requests does via the `proxies` parameter.
diff --git a/docs/async.md b/docs/async.md
index 45620bf..c631d54 100644
--- a/docs/async.md
+++ b/docs/async.md
@@ -105,3 +105,38 @@ trio.run(main)
 
 !!! important
     `trio` must be installed to import and use the `TrioBackend`.
+
+## FAQ
+
+### When should I use an `AsyncClient`?
+
+You should use an `AsyncClient` whenever you are inside an *async environment*.
+
+In particular, using `httpx.get()` or `httpx.Client()` in an async environment is **not** supported. There are several technical reasons to this, but the rationale is that you shouldn't be doing blocking-style network calls in the context of an event loop (for more discussion, see [#179](https://github.com/encode/httpx/issues/179)).
+
+For example, this won't work:
+
+```python
+import asyncio
+import httpx
+
+async def main():
+    r = httpx.get("https://example.org")
+
+asyncio.run(main())
+```
+
+Instead, you should use an `AsyncClient`:
+
+```python
+import asyncio
+import httpx
+
+async def main():
+    async with httpx.AsyncClient() as client:
+        r = await client.get("https://example.org")
+
+asyncio.run(main())
+```
+
+If you *need* to make synchronous requests, or otherwise run into issues related to sync usage, you should probably consider using a regular threaded client such as [Requests](https://github.com/psf/requests) instead.
diff --git a/docs/compatibility.md b/docs/compatibility.md
index 250a02c..7b2b391 100644
--- a/docs/compatibility.md
+++ b/docs/compatibility.md
@@ -32,3 +32,7 @@ with httpx.Client(**kwargs) as client:
 ```
 
 More detailed documentation and usage of `Client` can be found in [Advanced Usage](advanced.md).
+
+## Mocking
+
+If you need to mock HTTPX the same way that test utilities like `responses` and `requests-mock` does for `requests`, see [RESPX](https://github.com/lundberg/respx).
diff --git a/httpx/__version__.py b/httpx/__version__.py
index 5783d84..2f47ae7 100644
--- a/httpx/__version__.py
+++ b/httpx/__version__.py
@@ -1,3 +1,3 @@
 __title__ = "httpx"
 __description__ = "A next generation HTTP client, for Python 3."
-__version__ = "0.7.7"
+__version__ = "0.7.8"
diff --git a/httpx/client.py b/httpx/client.py
index 45589b9..c327c2d 100644
--- a/httpx/client.py
+++ b/httpx/client.py
@@ -74,6 +74,7 @@ class BaseClient:
         app: typing.Callable = None,
         backend: ConcurrencyBackend = None,
         trust_env: bool = True,
+        uds: str = None,
     ):
         if backend is None:
             backend = AsyncioBackend()
@@ -99,6 +100,7 @@ class BaseClient:
                 pool_limits=pool_limits,
                 backend=backend,
                 trust_env=self.trust_env,
+                uds=uds,
             )
         elif isinstance(dispatch, Dispatcher):
             async_dispatch = ThreadedDispatcher(dispatch, backend)
@@ -721,6 +723,7 @@ class Client(BaseClient):
     async requests.
     * **trust_env** - *(optional)* Enables or disables usage of environment
     variables for configuration.
+    * **uds** - *(optional)* A path to a Unix domain socket to connect through.
     """
 
     def check_concurrency_backend(self, backend: ConcurrencyBackend) -> None:
diff --git a/httpx/concurrency/asyncio.py b/httpx/concurrency/asyncio.py
index 010d821..a0163ed 100644
--- a/httpx/concurrency/asyncio.py
+++ b/httpx/concurrency/asyncio.py
@@ -58,16 +58,49 @@ class SocketStream(BaseSocketStream):
         self, hostname: str, ssl_context: ssl.SSLContext, timeout: TimeoutConfig
     ) -> "SocketStream":
         loop = asyncio.get_event_loop()
-        if not hasattr(loop, "start_tls"):  # pragma: no cover
-            raise NotImplementedError(
-                "asyncio.AbstractEventLoop.start_tls() is only available in Python 3.7+"
-            )
 
         stream_reader = asyncio.StreamReader()
         protocol = asyncio.StreamReaderProtocol(stream_reader)
         transport = self.stream_writer.transport
 
-        loop_start_tls = loop.start_tls  # type: ignore
+        if hasattr(loop, "start_tls"):
+            loop_start_tls = loop.start_tls  # type: ignore
+        else:
+
+            async def loop_start_tls(
+                transport: asyncio.BaseTransport,
+                protocol: asyncio.BaseProtocol,
+                sslcontext: ssl.SSLContext = None,
+                *,
+                server_side: bool = False,
+                server_hostname: str = None,
+                ssl_handshake_timeout: float = None,
+            ) -> asyncio.Transport:
+                """Python 3.6 asyncio doesn't have a start_tls() method on the loop
+                so we use this function in place of the loop's start_tls() method.
+                Adapted from this comment:
+                https://github.com/urllib3/urllib3/issues/1323#issuecomment-362494839
+                """
+                import asyncio.sslproto
+
+                waiter = loop.create_future()
+                ssl_protocol = asyncio.sslproto.SSLProtocol(
+                    loop,
+                    protocol,
+                    sslcontext,
+                    waiter,
+                    server_side=False,
+                    server_hostname=server_hostname,
+                    call_connection_made=False,
+                )
+
+                transport.set_protocol(ssl_protocol)
+                loop.call_soon(ssl_protocol.connection_made, transport)
+                loop.call_soon(transport.resume_reading)  # type: ignore
+
+                await waiter
+                return ssl_protocol._app_transport
+
         transport = await asyncio.wait_for(
             loop_start_tls(
                 transport=transport,
@@ -242,6 +275,29 @@ class AsyncioBackend(ConcurrencyBackend):
             stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout
         )
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> SocketStream:
+        server_hostname = hostname if ssl_context else None
+
+        try:
+            stream_reader, stream_writer = await asyncio.wait_for(  # type: ignore
+                asyncio.open_unix_connection(
+                    path, ssl=ssl_context, server_hostname=server_hostname
+                ),
+                timeout.connect_timeout,
+            )
+        except asyncio.TimeoutError:
+            raise ConnectTimeout()
+
+        return SocketStream(
+            stream_reader=stream_reader, stream_writer=stream_writer, timeout=timeout
+        )
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
diff --git a/httpx/concurrency/base.py b/httpx/concurrency/base.py
index 9d5bffd..2109c21 100644
--- a/httpx/concurrency/base.py
+++ b/httpx/concurrency/base.py
@@ -124,6 +124,15 @@ class ConcurrencyBackend:
     ) -> BaseSocketStream:
         raise NotImplementedError()  # pragma: no cover
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> BaseSocketStream:
+        raise NotImplementedError()  # pragma: no cover
+
     def get_semaphore(self, limits: PoolLimits) -> BasePoolSemaphore:
         raise NotImplementedError()  # pragma: no cover
 
diff --git a/httpx/concurrency/trio.py b/httpx/concurrency/trio.py
index 5d3b50d..c84b1e4 100644
--- a/httpx/concurrency/trio.py
+++ b/httpx/concurrency/trio.py
@@ -191,6 +191,26 @@ class TrioBackend(ConcurrencyBackend):
 
         return SocketStream(stream=stream, timeout=timeout)
 
+    async def open_uds_stream(
+        self,
+        path: str,
+        hostname: typing.Optional[str],
+        ssl_context: typing.Optional[ssl.SSLContext],
+        timeout: TimeoutConfig,
+    ) -> SocketStream:
+        connect_timeout = _or_inf(timeout.connect_timeout)
+
+        with trio.move_on_after(connect_timeout) as cancel_scope:
+            stream: trio.SocketStream = await trio.open_unix_socket(path)
+            if ssl_context is not None:
+                stream = trio.SSLStream(stream, ssl_context, server_hostname=hostname)
+                await stream.do_handshake()
+
+        if cancel_scope.cancelled_caught:
+            raise ConnectTimeout()
+
+        return SocketStream(stream=stream, timeout=timeout)
+
     async def run_in_threadpool(
         self, func: typing.Callable, *args: typing.Any, **kwargs: typing.Any
     ) -> typing.Any:
diff --git a/httpx/dispatch/connection.py b/httpx/dispatch/connection.py
index 91feb97..0612bcc 100644
--- a/httpx/dispatch/connection.py
+++ b/httpx/dispatch/connection.py
@@ -38,6 +38,7 @@ class HTTPConnection(AsyncDispatcher):
         http_versions: HTTPVersionTypes = None,
         backend: ConcurrencyBackend = None,
         release_func: typing.Optional[ReleaseCallback] = None,
+        uds: typing.Optional[str] = None,
     ):
         self.origin = Origin(origin) if isinstance(origin, str) else origin
         self.ssl = SSLConfig(cert=cert, verify=verify, trust_env=trust_env)
@@ -45,6 +46,7 @@ class HTTPConnection(AsyncDispatcher):
         self.http_versions = HTTPVersionConfig(http_versions)
         self.backend = AsyncioBackend() if backend is None else backend
         self.release_func = release_func
+        self.uds = uds
         self.h11_connection = None  # type: typing.Optional[HTTP11Connection]
         self.h2_connection = None  # type: typing.Optional[HTTP2Connection]
 
@@ -84,8 +86,21 @@ class HTTPConnection(AsyncDispatcher):
         else:
             on_release = functools.partial(self.release_func, self)
 
-        logger.trace(f"start_connect host={host!r} port={port!r} timeout={timeout!r}")
-        stream = await self.backend.open_tcp_stream(host, port, ssl_context, timeout)
+        if self.uds is None:
+            logger.trace(
+                f"start_connect tcp host={host!r} port={port!r} timeout={timeout!r}"
+            )
+            stream = await self.backend.open_tcp_stream(
+                host, port, ssl_context, timeout
+            )
+        else:
+            logger.trace(
+                f"start_connect uds path={self.uds!r} host={host!r} timeout={timeout!r}"
+            )
+            stream = await self.backend.open_uds_stream(
+                self.uds, host, ssl_context, timeout
+            )
+
         http_version = stream.get_http_version()
         logger.trace(f"connected http_version={http_version!r}")
 
diff --git a/httpx/dispatch/connection_pool.py b/httpx/dispatch/connection_pool.py
index 189fcff..5d7d886 100644
--- a/httpx/dispatch/connection_pool.py
+++ b/httpx/dispatch/connection_pool.py
@@ -89,6 +89,7 @@ class ConnectionPool(AsyncDispatcher):
         pool_limits: PoolLimits = DEFAULT_POOL_LIMITS,
         http_versions: HTTPVersionTypes = None,
         backend: ConcurrencyBackend = None,
+        uds: typing.Optional[str] = None,
     ):
         self.verify = verify
         self.cert = cert
@@ -97,6 +98,7 @@ class ConnectionPool(AsyncDispatcher):
         self.http_versions = http_versions
         self.is_closed = False
         self.trust_env = trust_env
+        self.uds = uds
 
         self.keepalive_connections = ConnectionStore()
         self.active_connections = ConnectionStore()
@@ -142,6 +144,7 @@ class ConnectionPool(AsyncDispatcher):
                 backend=self.backend,
                 release_func=self.release_connection,
                 trust_env=self.trust_env,
+                uds=self.uds,
             )
             logger.trace(f"new_connection connection={connection!r}")
         else:


Test patch:
diff --git a/tests/client/test_async_client.py b/tests/client/test_async_client.py
index eaac5ef..42202aa 100644
--- a/tests/client/test_async_client.py
+++ b/tests/client/test_async_client.py
@@ -146,3 +146,14 @@ async def test_100_continue(server, backend):
 
     assert response.status_code == 200
     assert response.content == data
+
+
+async def test_uds(uds_server, backend):
+    url = uds_server.url
+    uds = uds_server.config.uds
+    assert uds is not None
+    async with httpx.AsyncClient(backend=backend, uds=uds) as client:
+        response = await client.get(url)
+    assert response.status_code == 200
+    assert response.text == "Hello, world!"
+    assert response.encoding == "iso-8859-1"
diff --git a/tests/client/test_client.py b/tests/client/test_client.py
index f7be607..5dc196d 100644
--- a/tests/client/test_client.py
+++ b/tests/client/test_client.py
@@ -138,6 +138,17 @@ def test_base_url(server):
     assert response.url == base_url
 
 
+def test_uds(uds_server):
+    url = uds_server.url
+    uds = uds_server.config.uds
+    assert uds is not None
+    with httpx.Client(uds=uds) as http:
+        response = http.get(url)
+    assert response.status_code == 200
+    assert response.text == "Hello, world!"
+    assert response.encoding == "iso-8859-1"
+
+
 def test_merge_url():
     client = httpx.Client(base_url="https://www.paypal.com/")
     url = client.merge_url("http://www.paypal.com")
diff --git a/tests/conftest.py b/tests/conftest.py
index ef57cae..de67ff7 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -288,6 +288,15 @@ def server():
     yield from serve_in_thread(server)
 
 
+@pytest.fixture(scope=SERVER_SCOPE)
+def uds_server():
+    uds = "test_server.sock"
+    config = Config(app=app, lifespan="off", loop="asyncio", uds=uds)
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+    os.remove(uds)
+
+
 @pytest.fixture(scope=SERVER_SCOPE)
 def https_server(cert_pem_file, cert_private_key_file):
     config = Config(
@@ -301,3 +310,19 @@ def https_server(cert_pem_file, cert_private_key_file):
     )
     server = TestServer(config=config)
     yield from serve_in_thread(server)
+
+
+@pytest.fixture(scope=SERVER_SCOPE)
+def https_uds_server(cert_pem_file, cert_private_key_file):
+    uds = "https_test_server.sock"
+    config = Config(
+        app=app,
+        lifespan="off",
+        ssl_certfile=cert_pem_file,
+        ssl_keyfile=cert_private_key_file,
+        uds=uds,
+        loop="asyncio",
+    )
+    server = TestServer(config=config)
+    yield from serve_in_thread(server)
+    os.remove(uds)
diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py
index 8bb933b..cf3844c 100644
--- a/tests/test_concurrency.py
+++ b/tests/test_concurrency.py
@@ -1,5 +1,3 @@
-import sys
-
 import pytest
 import trio
 
@@ -26,20 +24,26 @@ from httpx.concurrency.trio import TrioBackend
         ),
     ],
 )
-async def test_start_tls_on_socket_stream(https_server, backend, get_cipher):
+@pytest.mark.parametrize("use_uds", (False, True))
+async def test_start_tls_on_socket_stream(
+    https_server, https_uds_server, backend, get_cipher, use_uds
+):
     """
     See that the concurrency backend can make a connection without TLS then
     start TLS on an existing connection.
     """
-    if isinstance(backend, AsyncioBackend) and sys.version_info < (3, 7):
-        pytest.xfail(reason="Requires Python 3.7+ for AbstractEventLoop.start_tls()")
-
     ctx = SSLConfig().load_ssl_context_no_verify(HTTPVersionConfig())
     timeout = TimeoutConfig(5)
 
-    stream = await backend.open_tcp_stream(
-        https_server.url.host, https_server.url.port, None, timeout
-    )
+    if use_uds:
+        assert https_uds_server.config.uds is not None
+        stream = await backend.open_uds_stream(
+            https_uds_server.config.uds, https_uds_server.url.host, None, timeout
+        )
+    else:
+        stream = await backend.open_tcp_stream(
+            https_server.url.host, https_server.url.port, None, timeout
+        )
 
     try:
         assert stream.is_connection_dropped() is False