metered_session.py

class MeteredSession(requests.Session):
    """
    Creates a session where response downloads are shown graphically on the
    terminal with `click.progressbar` meters. This session adds a hook to
    responses that will attach 2 meter methods, described below.

    To use these meter methods on a response obj `resp`, call
    * `resp.iter_content_metered(chunk_size, label='\t' + resp.url)`
    * `resp.consume_content_metered(chunk_size, label='\t' + resp.url)`

    Note: `stream` is set to true for these responses because it must be.
    Otherwise, the meter will show just the in-memory iteration over the content
    chunks that have already been downloaded. When stream is True, each iteration
    corresponds to a read from the socket (...sort of, there's prolly buffer
    size issues etc).

    Fuck dude, i don't know this really works as described after reading
    requests/models.py.
    """
    def __init__(self, decode_unicode=False, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.hooks.update({'response': MeteredSession.attach_meter_methods})
        self.stream = True

    @staticmethod
    def attach_meter_methods(resp, *args, **kwargs):
        build_iter_content_metered(resp)
        build_content_metered(resp)

    @staticmethod
    def build_iter_content_metered(resp):
        def iter_content_metered(chunk_size, decode_unicode=False,
            label='\t' + resp.url):
            """Like requests' `response.iter_content`, but output a progress bar
            meter to the terminal.
            """
            length = int(resp.headers['content-length'])/chunk_size or None
            with click.progressbar(resp.iter_content(chunk_size),
                                   label=label,
                                   length=length) as meter:
                for chunk in meter:
                    yield chunk
        resp.iter_content_metered = iter_content_metered

    @staticmethod
    def build_content_metered(resp):
        def content_metered(chunk_size, decode_unicode=False,
            label='\t' + resp.url):
            # content =
            for chunk in iter_content_metered(chunk_size, decode_unicode, label):
                pass
        resp.consume_content_metered = consume_content_metered