nonebot2/nonebot/drivers/httpx.py

80 lines
1.8 KiB
Python
Raw Normal View History

2022-01-22 15:23:07 +08:00
"""[HTTPX](https://www.python-httpx.org/) 驱动适配
```bash
nb driver install httpx
# 或者
pip install nonebot2[httpx]
```
:::tip 提示
本驱动仅支持客户端 HTTP 连接
:::
FrontMatter:
sidebar_position: 3
description: nonebot.drivers.httpx 模块
"""
from typing import TYPE_CHECKING
from typing_extensions import override
from nonebot.drivers.none import Driver as NoneDriver
2021-12-22 16:53:55 +08:00
from nonebot.drivers import (
Request,
Response,
HTTPVersion,
HTTPClientMixin,
2021-12-22 16:53:55 +08:00
combine_driver,
)
try:
import httpx
except ModuleNotFoundError as e: # pragma: no cover
2021-12-25 14:04:53 +08:00
raise ImportError(
"Please install httpx first to use this driver. "
"Install with pip: `pip install nonebot2[httpx]`"
) from e
2021-12-22 16:53:55 +08:00
class Mixin(HTTPClientMixin):
2022-01-22 15:23:07 +08:00
"""HTTPX Mixin"""
2021-12-22 16:53:55 +08:00
@property
@override
2021-12-22 16:53:55 +08:00
def type(self) -> str:
return "httpx"
@override
2021-12-22 16:53:55 +08:00
async def request(self, setup: Request) -> Response:
async with httpx.AsyncClient(
cookies=setup.cookies.jar,
2021-12-25 14:04:53 +08:00
http2=setup.version == HTTPVersion.H2,
proxies=setup.proxy,
2021-12-25 14:04:53 +08:00
follow_redirects=True,
2021-12-22 16:53:55 +08:00
) as client:
response = await client.request(
setup.method,
str(setup.url),
content=setup.content,
data=setup.data,
json=setup.json,
files=setup.files,
2021-12-22 16:53:55 +08:00
headers=tuple(setup.headers.items()),
timeout=setup.timeout,
)
return Response(
response.status_code,
headers=response.headers.multi_items(),
2021-12-22 16:53:55 +08:00
content=response.content,
request=setup,
)
if TYPE_CHECKING:
class Driver(Mixin, NoneDriver): ...
2021-12-22 16:53:55 +08:00
else:
Driver = combine_driver(NoneDriver, Mixin)
"""HTTPX Driver"""