Exposing the content stream API

[tomchristie]

While we've pretty much got all of our public/private API properly demarcated now, there's one aspect where that's still not quite true, which is the content stream API. We currently expose ContentStream parameters or properties in the following places...

• Request(..., stream=<ContentStream>) - Exposed in order to copy + modify a request eg. on redirect or digest auth.
• Response(..., stream=<ContentStream>) - Exposed in order to pass from the transport layer when constructing the response.
• request.stream - Exposed in order to pass to the transport layer when sending the request, or to copy + modify a request eg. on redirect or digest auth.

We're also a bit wooly about if we actually mean an httpcore.AsyncByteStream/httpcore.SyncByteStream or an httpx.ContentStream in places where that's exposed. Furthermore, we actually would like to expose content streams for some advanced functionality, like being able to support streaming the content of a file, while providing replay support.

Eg...

def upload_file(name):
    with open(name, "rb") as input_file:
        while True:
            chunk = input_file.read(4096)
            if not chunk:
                break
            yield chunk

# This will fail on some kinds of redirect, or on digest auth
httpx.post(data=upload_file("example.zip"))

But if we exposed the content stream API then we could support something like...

class UploadFile(httpx.ContentStream):
    def __init__(self, name):
        self.name = name

    def __iter__(self):
        with open(self.name, "rb") as input_file:
            while True:
                chunk = input_file.read(4096)
                if not chunk:
                    break
                yield chunk

# This can replay, if needed
httpx.post(data=UploadFile("example.zip"))

Because we're a bit fuzzy around the ContentStream vs. httpcore.AsyncByteStream/httpcore.SyncByteStream I think we also need an API that's more clear there, and doesn't have this odd artifact of .can_replay() and .get_headers() methods, that are used in some contexts, but not in others.

Here's what I think we need to do here.

• Change the httpcore API, so that we only have httpcore.ByteStream, which is an abstract base class with for a unified sync+async byte stream. Ie. it exposes __iter__, __aiter__, close, aclose methods.
• Drop can_replay() on the httpx implementations. Instead raise an exception in __iter__/__aiter__ if we attempt to replay a generator byte stream. Eg. StreamConsumed("Attempted to iterate over a byte stream, but the stream has already been consumed.") If necessary we could catch that in auth/redirect contexts and provide more information about exactly why it's occured.
• Drop get_headers() on the httpx implementations, and instead have encode return a two-tuple of (headers, stream).
• Use the naming ByteStream instead of ContentStream.

I guess httpx.ByteStream would then just be a synonym for httpcore.ByteStream.

Also, the only byte stream class that we would expose would be the abstract base httpx.ByteStream - If you want to use it you have to subclass it and provide a concrete implementation. For anything else you're already covered by the standard data/files/json types.

[florimondmanca]

Delightful write up Tom, as always!

Sounds like a nice way to expose the API while not exposing the entire hierarchy of actual implementations (which I agree ought to remain private at this time).

Some actionable items already, though I'm not sure I wrapped my head around all possible details, but we can discuss those against a PR. 👍

[tomchristie]

Thanks @florimondmanca - main thing here is just to try to explain what needs to change and why, before we expose an httpx.ByteStream interface.

I'll aim address this in step by step PRs, and make the motivation clear at each point.
We'll end up with work against each of the following...

• Introduce an httpcore.ByteStream class, instead of httpcore.AsyncByteStream/httpcore.SyncByteStream
• Support Request(content=<byte stream>) and Response(content=<byte stream>), rather than having a dedicated stream= argument.
• Expose response.stream in parallel to the existing request.stream.
• Refactoring to trim down the existing ContentStream class into just a plain ByteStream class that we can expose as public API.

[tomchristie]

Actually we may not need to merge httpcore.AsyncByteStream/httpcore.SyncByteStream in order to get where we're going here.

We can do this perfectly well with the existing httpcore.SyncByteStream/httpcore.AsyncByteStream interfaces, and end up with...

• Request(..., content=<bytes|byte iterator|byte aiterator|SyncByteStream|AsyncByteStream)
• Response(..., content=<bytes|byte iterator|byte aiterator|SyncByteStream|AsyncByteStream)
• Request.stream <SyncByteStream|AsyncByteStream>
• Response.stream <SyncByteStream|AsyncByteStream>

We still need the refactoring to drop .can_replay and .get_headers() and remove the extra ContentStream abstraction.