Expose network backends as part of the public API. #695

import httpcore
from httpcore.backends.base import NetworkBackend, NetworkStream
from httpcore.backends.sync import SyncBackend


class RecordingNetworkStream(NetworkStream):
    def __init__(self, record_file, stream):
        self.record_file = record_file
        self.stream = stream

    def read(self, max_bytes, timeout=None):
        data = self.stream.read(max_bytes, timeout=timeout)
        self.record_file.write(data)
        return data

    def write(self, buffer, timeout=None):
        self.stream.write(buffer, timeout=timeout)

    def close(self) -> None:
        self.stream.close()

    def start_tls(
        self,
        ssl_context,
        server_hostname=None,
        timeout=None,
    ):
        self.stream = self.stream.start_tls(
            ssl_context, server_hostname=server_hostname, timeout=timeout
        )
        return self

    def get_extra_info(self, info):
        return self.stream.get_extra_info(info)


class RecordingNetworkBackend(NetworkBackend):
    def __init__(self, record_file):
        self.record_file = record_file
        self.backend = SyncBackend()

    def connect_tcp(
        self,
        host,
        port,
        timeout=None,
        local_address=None,
    ) -> NetworkStream:
        stream = self.backend.connect_tcp(
            host, port, timeout=timeout, local_address=local_address
        )
        return RecordingNetworkStream(self.record_file, stream)

    def connect_unix_socket(self, path, timeout=None) -> NetworkStream:
        stream = self.backend.connect_unix_socket(path, timeout=timeout)
        return RecordingNetworkStream(self.record_file, stream)

    def sleep(self, seconds: float) -> None:
        self.backend.sleep(seconds)


with open("network-recording", "wb") as record_file:
    network_backend = RecordingNetworkBackend(record_file)
    with httpcore.ConnectionPool(network_backend=network_backend) as http:
        http.request("GET", "https://www.example.com/")