Python aiohttp 模块，HttpProcessingError() 实例源码

我们从Python开源项目中，提取了以下9个代码示例，用于说明如何使用aiohttp.HttpProcessingError()。

项目：notebooks 作者：fluentpython | 项目源码 | 文件源码

def get_flag(base_url, cc):
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    with contextlib.closing(resp):
        if resp.status == 200:
            image = yield from resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)


# BEGIN FLAGS2_ASYNCIO_EXECUTOR

项目：client 作者：syncrypt | 项目源码 | 文件源码

def read(self, count=-1):
        if self._done:
            return b''
        if self.response is None:
            self.response = yield from self.client.put(self.url,
                    data=self.feed_http_upload(),
                    headers={} if self.size is None else {'Content-Length': str(self.size)})
        content = yield from self.response.read()
        yield from self.response.release()
        if not self.response.status in (200, 201, 202):
            raise aiohttp.HttpProcessingError(
                code=self.response.status, message=self.response.reason,
                headers=self.response.headers)
        self._done = True
        if 'ETAG' in self.response.headers:
            self.etag = self.response.headers['ETAG'][1:-1]
        return content

项目：Books_SourceCode 作者：activeion | 项目源码 | 文件源码

def get_flag(base_url, cc):
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    with contextlib.closing(resp):
        if resp.status == 200:
            image = yield from resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)


# BEGIN FLAGS2_ASYNCIO_EXECUTOR

项目：Books_SourceCode 作者：activeion | 项目源码 | 文件源码

def get_flag(base_url, cc):
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    with contextlib.closing(resp):
        if resp.status == 200:
            image = yield from resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)


# BEGIN FLAGS2_ASYNCIO_EXECUTOR

项目：notebooks 作者：fluentpython | 项目源码 | 文件源码

def get_flag(base_url, cc): # <2>
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    with contextlib.closing(resp):
        if resp.status == 200:
            image = yield from resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)

项目：notebooks 作者：fluentpython | 项目源码 | 文件源码

def get_flag(base_url, cc): # <2>
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    with closing(await aiohttp.request('GET', url)) as resp:
        if resp.status == 200:
            image = await resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)

项目：zenchmarks 作者：squeaky-pl | 项目源码 | 文件源码

def raise_for_status(self):
        if 400 <= self.status:
            raise aiohttp.HttpProcessingError(
                code=self.status,
                message=self.reason)

项目：Books_SourceCode 作者：activeion | 项目源码 | 文件源码

def get_flag(base_url, cc): # <2>
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    resp = yield from aiohttp.request('GET', url)
    with contextlib.closing(resp):
        if resp.status == 200:
            image = yield from resp.read()
            return image
        elif resp.status == 404:
            raise web.HTTPNotFound()
        else:
            raise aiohttp.HttpProcessingError(
                code=resp.status, message=resp.reason,
                headers=resp.headers)

项目：Books_SourceCode 作者：activeion | 项目源码 | 文件源码

def get_flag(base_url, cc): # <2>
    url = '{}/{cc}/{cc}.gif'.format(base_url, cc=cc.lower())
    async with aiohttp.ClientSession() as session:         
        with async_timeout.timeout(10):         
            async with session.get(url) as resp:             
                if resp.status == 200:
                    image = await resp.read()  # <5>
                    return image
                elif resp.status == 404:
                    raise web.HTTPNotFound()
                else:
                    raise aiohttp.HttpProcessingError(
                        code=resp.status, message=resp.reason,
                        headers=resp.headers)