nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-03-20 03:57:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

添加异步请求工具类；优化fetch_image和proxy_img函数为异步实现提升性能

Browse Source
This commit is contained in:
jxxghp
2025-07-30 08:30:24 +08:00
parent 2428d58e93
commit f077a9684b
4 changed files with 463 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/api/endpoints/login.py
View File

@@ -44,7 +44,7 @@ def login_access_token(
user_name=user_or_message.name,
avatar=user_or_message.avatar,
level=level,
permissions= user_or_message.permissions or {},
permissions=user_or_message.permissions or {},
)

35
app/api/endpoints/system.py
View File

@@ -2,10 +2,8 @@ import asyncio
import io
import json
import re
import tempfile
from collections import deque
from datetime import datetime
from pathlib import Path
from typing import Optional, Union, Annotated
import aiofiles
@@ -37,7 +35,7 @@ from app.scheduler import Scheduler
from app.schemas import ConfigChangeEventData
from app.schemas.types import SystemConfigKey, EventType
from app.utils.crypto import HashUtils
from app.utils.http import RequestUtils
from app.utils.http import RequestUtils, AsyncRequestUtils
from app.utils.security import SecurityUtils
from app.utils.url import UrlUtils
from version import APP_VERSION
@@ -45,7 +43,7 @@ from version import APP_VERSION
router = APIRouter()
def fetch_image(
async def fetch_image(
url: str,
proxy: bool = False,
use_disk_cache: bool = False,
@@ -82,7 +80,8 @@ def fetch_image(
# 目前暂不考虑磁盘缓存文件是否过期，后续通过缓存清理机制处理
if cache_path.exists():
try:
content = cache_path.read_bytes()
async with aiofiles.open(cache_path, 'rb') as f:
content = await f.read()
etag = HashUtils.md5(content)
headers = RequestUtils.generate_cache_headers(etag, max_age=86400 * 7)
if if_none_match == etag:
@@ -95,19 +94,19 @@ def fetch_image(
# 请求远程图片
referer = "https://movie.douban.com/" if "doubanio.com" in url else None
proxies = settings.PROXY if proxy else None
response = RequestUtils(ua=settings.NORMAL_USER_AGENT, proxies=proxies, referer=referer,
accept_type="image/avif,image/webp,image/apng,*/*").get_res(url=url)
response = await AsyncRequestUtils(ua=settings.NORMAL_USER_AGENT, proxies=proxies, referer=referer,
accept_type="image/avif,image/webp,image/apng,*/*").get_res(url=url)
if not response:
raise HTTPException(status_code=502, detail="Failed to fetch the image from the remote server")
# 验证下载的内容是否为有效图片
try:
Image.open(io.BytesIO(response.content)).verify()
content = response.content
Image.open(io.BytesIO(content)).verify()
except Exception as e:
logger.debug(f"Invalid image format for URL {url}: {e}")
raise HTTPException(status_code=502, detail="Invalid image format")
content = response.content
response_headers = response.headers
cache_control_header = response_headers.get("Cache-Control", "")
@@ -118,10 +117,9 @@ def fetch_image(
try:
if not cache_path.parent.exists():
cache_path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile(dir=cache_path.parent, delete=False) as tmp_file:
tmp_file.write(content)
temp_path = Path(tmp_file.name)
temp_path.replace(cache_path)
# 使用异步文件操作写入缓存
async with aiofiles.open(cache_path, 'wb') as f:
await f.write(content)
except Exception as e:
logger.debug(f"Failed to write cache file {cache_path}: {e}")
@@ -141,7 +139,7 @@ def fetch_image(
@router.get("/img/{proxy}", summary="图片代理")
def proxy_img(
async def proxy_img(
imgurl: str,
proxy: bool = False,
cache: bool = False,
@@ -155,12 +153,12 @@ def proxy_img(
hosts = [config.config.get("host") for config in MediaServerHelper().get_configs().values() if
config and config.config and config.config.get("host")]
allowed_domains = set(settings.SECURITY_IMAGE_DOMAINS) | set(hosts)
return fetch_image(url=imgurl, proxy=proxy, use_disk_cache=cache,
if_none_match=if_none_match, allowed_domains=allowed_domains)
return await fetch_image(url=imgurl, proxy=proxy, use_disk_cache=cache,
if_none_match=if_none_match, allowed_domains=allowed_domains)
@router.get("/cache/image", summary="图片缓存")
def cache_img(
async def cache_img(
url: str,
if_none_match: Annotated[str | None, Header()] = None,
_: schemas.TokenPayload = Depends(verify_resource_token)
@@ -170,7 +168,8 @@ def cache_img(
"""
# 如果没有启用全局图片缓存，则不使用磁盘缓存
proxy = "doubanio.com" not in url
return fetch_image(url=url, proxy=proxy, use_disk_cache=settings.GLOBAL_IMAGE_CACHE, if_none_match=if_none_match)
return await fetch_image(url=url, proxy=proxy, use_disk_cache=settings.GLOBAL_IMAGE_CACHE,
if_none_match=if_none_match)
@router.get("/global", summary="查询非敏感系统设置", response_model=schemas.Response)

640
app/utils/http.py
View File

@@ -1,10 +1,11 @@
import sys
import re
from contextlib import contextmanager
import sys
from contextlib import contextmanager, asynccontextmanager
from pathlib import Path
from typing import Any, Optional, Union
import chardet
import httpx
import requests
import urllib3
from requests import Response, Session
@@ -16,81 +17,67 @@ from app.log import logger
urllib3.disable_warnings(InsecureRequestWarning)
class AutoCloseResponse:
def cookie_parse(cookies_str: str, array: bool = False) -> Union[list, dict]:
"""
自动关闭连接的Response包装器
在访问常用属性后自动关闭连接
解析cookie，转化为字典或者数组
:param cookies_str: cookie字符串
:param array: 是否转化为数组
:return: 字典或者数组
"""
if not cookies_str:
return {}
cookie_dict = {}
cookies = cookies_str.split(";")
for cookie in cookies:
cstr = cookie.split("=")
if len(cstr) > 1:
cookie_dict[cstr[0].strip()] = cstr[1].strip()
if array:
return [{"name": k, "value": v} for k, v in cookie_dict.items()]
return cookie_dict
def __init__(self, response: Response):
self._response = response
self._closed = False
def __getattr__(self, name):
"""
对于其他属性，直接委托给原始response
"""
return getattr(self._response, name)
def get_caller():
"""
获取调用者的名称，识别是否为插件调用
"""
# 调用者名称
caller_name = None
def _auto_close(self):
"""
自动关闭连接
"""
if not self._closed and self._response:
try:
self._response.close()
self._closed = True
except Exception as e:
logger.debug(f"自动关闭响应失败: {e}")
try:
frame = sys._getframe(3) # noqa
except (AttributeError, ValueError):
return None
def json(self, **kwargs):
"""
获取JSON数据并自动关闭连接
"""
while frame:
filepath = Path(frame.f_code.co_filename)
parts = filepath.parts
if "app" in parts:
if not caller_name and "plugins" in parts:
try:
plugins_index = parts.index("plugins")
if plugins_index + 1 < len(parts):
plugin_candidate = parts[plugins_index + 1]
if plugin_candidate != "__init__.py":
caller_name = plugin_candidate
break
except ValueError:
pass
if "main.py" in parts:
break
elif len(parts) != 1:
break
try:
data = self._response.json(**kwargs)
return data
finally:
self._auto_close()
@property
def text(self):
"""
获取文本内容并自动关闭连接
"""
try:
return self._response.text
finally:
self._auto_close()
@property
def content(self):
"""
获取二进制内容并自动关闭连接
"""
try:
return self._response.content
finally:
self._auto_close()
def close(self):
"""
手动关闭连接
"""
self._auto_close()
def __setstate__(self, state):
for name, value in state.items():
setattr(self, name, value)
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
frame = frame.f_back
except AttributeError:
break
return caller_name
class RequestUtils:
"""
HTTP请求工具类，提供同步HTTP请求的基本功能
"""
def __init__(self,
headers: dict = None,
@@ -102,6 +89,17 @@ class RequestUtils:
referer: str = None,
content_type: str = None,
accept_type: str = None):
"""
:param headers: 请求头部信息
:param ua: User-Agent字符串
:param cookies: Cookie字符串或字典
:param proxies: 代理设置
:param session: requests.Session实例，如果为None则创建新的Session
:param timeout: 请求超时时间，默认为20秒
:param referer: Referer头部信息
:param content_type: 请求的Content-Type，默认为 "application/x-www-form-urlencoded; charset=UTF-8"
:param accept_type: Accept头部信息，默认为 "application/json"
"""
self._proxies = proxies
self._session = session
self._timeout = timeout or 20
@@ -111,7 +109,7 @@ class RequestUtils:
self._headers = headers
else:
if ua and ua == settings.USER_AGENT:
caller_name = self.__get_caller()
caller_name = get_caller()
if caller_name:
ua = f"{settings.USER_AGENT} Plugin/{caller_name}"
self._headers = {
@@ -122,48 +120,30 @@ class RequestUtils:
}
if cookies:
if isinstance(cookies, str):
self._cookies = self.cookie_parse(cookies)
self._cookies = cookie_parse(cookies)
else:
self._cookies = cookies
else:
self._cookies = None
@staticmethod
def __get_caller():
@contextmanager
def response_manager(self, method: str, url: str, **kwargs):
"""
获取调用者的名称，识别是否为插件调用
响应管理器上下文管理器，确保响应对象被正确关闭
:param method: HTTP方法
:param url: 请求的URL
:param kwargs: 其他请求参数
"""
# 调用者名称
caller_name = None
response = None
try:
frame = sys._getframe(3) # noqa
except (AttributeError, ValueError):
return None
while frame:
filepath = Path(frame.f_code.co_filename)
parts = filepath.parts
if "app" in parts:
if not caller_name and "plugins" in parts:
try:
plugins_index = parts.index("plugins")
if plugins_index + 1 < len(parts):
plugin_candidate = parts[plugins_index + 1]
if plugin_candidate != "__init__.py":
caller_name = plugin_candidate
break
except ValueError:
pass
if "main.py" in parts:
break
elif len(parts) != 1:
break
try:
frame = frame.f_back
except AttributeError:
break
return caller_name
response = self.request(method=method, url=url, **kwargs)
yield response
finally:
if response:
try:
response.close()
except Exception as e:
logger.debug(f"关闭响应失败: {e}")
def request(self, method: str, url: str, raise_exception: bool = False, **kwargs) -> Optional[Response]:
"""
@@ -210,7 +190,7 @@ class RequestUtils:
logger.debug(f"处理响应内容失败: {e}")
return None
finally:
response.close() # 确保连接被关闭
response.close()
return None
def post(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[Response]:
@@ -222,8 +202,6 @@ class RequestUtils:
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestException则返回None
"""
if json is None:
json = {}
return self.request(method="post", url=url, data=data, json=json, **kwargs)
def put(self, url: str, data: Any = None, **kwargs) -> Optional[Response]:
@@ -243,8 +221,7 @@ class RequestUtils:
json: dict = None,
allow_redirects: bool = True,
raise_exception: bool = False,
auto_close: bool = True,
**kwargs) -> Optional[AutoCloseResponse]:
**kwargs) -> Optional[Response]:
"""
发送GET请求并返回响应对象
:param url: 请求的URL
@@ -253,22 +230,18 @@ class RequestUtils:
:param json: 请求的JSON数据
:param allow_redirects: 是否允许重定向
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param auto_close: 是否自动关闭响应连接，None时使用全局配置
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestException则返回None
:raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
"""
response = self.request(method="get",
url=url,
params=params,
data=data,
json=json,
allow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
if response is not None and auto_close:
return AutoCloseResponse(response)
return response
return self.request(method="get",
url=url,
params=params,
data=data,
json=json,
allow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
@contextmanager
def get_stream(self, url: str, params: dict = None, **kwargs):
@@ -294,8 +267,7 @@ class RequestUtils:
files: Any = None,
json: dict = None,
raise_exception: bool = False,
auto_close: bool = True,
**kwargs) -> Optional[AutoCloseResponse]:
**kwargs) -> Optional[Response]:
"""
发送POST请求并返回响应对象
:param url: 请求的URL
@@ -305,23 +277,19 @@ class RequestUtils:
:param files: 请求的文件
:param json: 请求的JSON数据
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param auto_close: 是否自动关闭响应连接，None时使用全局配置
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestException则返回None
:raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
"""
response = self.request(method="post",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
if response is not None and auto_close:
return AutoCloseResponse(response)
return response
return self.request(method="post",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
def put_res(self,
url: str,
@@ -331,8 +299,7 @@ class RequestUtils:
files: Any = None,
json: dict = None,
raise_exception: bool = False,
auto_close: bool = True,
**kwargs) -> Optional[AutoCloseResponse]:
**kwargs) -> Optional[Response]:
"""
发送PUT请求并返回响应对象
:param url: 请求的URL
@@ -342,23 +309,19 @@ class RequestUtils:
:param files: 请求的文件
:param json: 请求的JSON数据
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param auto_close: 是否自动关闭响应连接，None时使用全局配置
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestException则返回None
:raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
"""
response = self.request(method="put",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
if response is not None and auto_close:
return AutoCloseResponse(response)
return response
return self.request(method="put",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
def delete_res(self,
url: str,
@@ -366,8 +329,7 @@ class RequestUtils:
params: dict = None,
allow_redirects: bool = True,
raise_exception: bool = False,
auto_close: bool = True,
**kwargs) -> Optional[AutoCloseResponse]:
**kwargs) -> Optional[Response]:
"""
发送DELETE请求并返回响应对象
:param url: 请求的URL
@@ -375,41 +337,60 @@ class RequestUtils:
:param params: 请求的参数
:param allow_redirects: 是否允许重定向
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param auto_close: 是否自动关闭响应连接，None时使用全局配置
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestException则返回None
:raises: requests.exceptions.RequestException 仅raise_exception为True时会抛出
"""
response = self.request(method="delete",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
if response is not None and auto_close:
return AutoCloseResponse(response)
return response
return self.request(method="delete",
url=url,
data=data,
params=params,
allow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
@staticmethod
def cookie_parse(cookies_str: str, array: bool = False) -> Union[list, dict]:
def get_json(self, url: str, params: dict = None, **kwargs) -> Optional[dict]:
"""
解析cookie，转化为字典或者数组
:param cookies_str: cookie字符串
:param array: 是否转化为数组
:return: 字典或者数组
发送GET请求并返回JSON数据，自动关闭连接
:param url: 请求的URL
:param params: 请求的参数
:param kwargs: 其他请求参数
:return: JSON数据，若发生异常则返回None
"""
if not cookies_str:
return {}
cookie_dict = {}
cookies = cookies_str.split(";")
for cookie in cookies:
cstr = cookie.split("=")
if len(cstr) > 1:
cookie_dict[cstr[0].strip()] = cstr[1].strip()
if array:
return [{"name": k, "value": v} for k, v in cookie_dict.items()]
return cookie_dict
response = self.request(method="get", url=url, params=params, **kwargs)
if response:
try:
data = response.json()
return data
except Exception as e:
logger.debug(f"解析JSON失败: {e}")
return None
finally:
response.close()
return None
def post_json(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[dict]:
"""
发送POST请求并返回JSON数据，自动关闭连接
:param url: 请求的URL
:param data: 请求的数据
:param json: 请求的JSON数据
:param kwargs: 其他请求参数
:return: JSON数据，若发生异常则返回None
"""
if json is None:
json = {}
response = self.request(method="post", url=url, data=data, json=json, **kwargs)
if response:
try:
data = response.json()
return data
except Exception as e:
logger.debug(f"解析JSON失败: {e}")
return None
finally:
response.close()
return None
@staticmethod
def parse_cache_control(header: str) -> (str, int):
@@ -522,7 +503,7 @@ class RequestUtils:
return fallback_encoding or "utf-8"
@staticmethod
def get_decoded_html_content(response: Union[Response, AutoCloseResponse],
def get_decoded_html_content(response: Response,
performance_mode: bool = False, confidence_threshold: float = 0.8) -> str:
"""
获取HTML响应的解码文本内容
@@ -555,48 +536,315 @@ class RequestUtils:
logger.debug(f"Error when getting decoded content: {str(e)}")
return response.text
@contextmanager
def response_manager(self, method: str, url: str, **kwargs):
class AsyncRequestUtils:
"""
异步HTTP请求工具类，提供异步HTTP请求的基本功能
"""
def __init__(self,
headers: dict = None,
ua: str = None,
cookies: Union[str, dict] = None,
proxies: dict = None,
client: httpx.AsyncClient = None,
timeout: int = None,
referer: str = None,
content_type: str = None,
accept_type: str = None):
"""
响应管理器上下文管理器，确保响应对象被正确关闭
:param headers: 请求头部信息
:param ua: User-Agent字符串
:param cookies: Cookie字符串或字典
:param proxies: 代理设置
:param client: httpx.AsyncClient实例，如果为None则创建新的客户端
:param timeout: 请求超时时间，默认为20秒
:param referer: Referer头部信息
:param content_type: 请求的Content-Type，默认为 "application/x-www-form-urlencoded; charset=UTF-8"
:param accept_type: Accept头部信息，默认为 "application/json"
"""
self._proxies = proxies
self._client = client
self._timeout = timeout or 20
if not content_type:
content_type = "application/x-www-form-urlencoded; charset=UTF-8"
if headers:
self._headers = headers
else:
if ua and ua == settings.USER_AGENT:
caller_name = get_caller()
if caller_name:
ua = f"{settings.USER_AGENT} Plugin/{caller_name}"
self._headers = {
"User-Agent": ua,
"Content-Type": content_type,
"Accept": accept_type,
"referer": referer
}
if cookies:
if isinstance(cookies, str):
self._cookies = cookie_parse(cookies)
else:
self._cookies = cookies
else:
self._cookies = None
@asynccontextmanager
async def response_manager(self, method: str, url: str, **kwargs):
"""
异步响应管理器上下文管理器，确保响应对象被正确关闭
:param method: HTTP方法
:param url: 请求的URL
:param kwargs: 其他请求参数
"""
response = None
try:
response = self.request(method=method, url=url, **kwargs)
response = await self.request(method=method, url=url, **kwargs)
yield response
finally:
if response:
try:
response.close()
await response.aclose()
except Exception as e:
logger.debug(f"关闭响应失败: {e}")
logger.debug(f"关闭异步响应失败: {e}")
def get_json(self, url: str, params: dict = None, **kwargs) -> Optional[dict]:
async def request(self, method: str, url: str, raise_exception: bool = False, **kwargs) -> Optional[httpx.Response]:
"""
发送GET请求并返回JSON数据，自动关闭连接
发起异步HTTP请求
:param method: HTTP方法，如 get, post, put 等
:param url: 请求的URL
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象
:raises: httpx.RequestError 仅raise_exception为True时会抛出
"""
if self._client is None:
# 创建临时客户端
async with httpx.AsyncClient(
proxy=self._proxies,
timeout=self._timeout,
verify=False,
follow_redirects=True
) as client:
return await self._make_request(client, method, url, raise_exception, **kwargs)
else:
return await self._make_request(self._client, method, url, raise_exception, **kwargs)
async def _make_request(self, client: httpx.AsyncClient, method: str, url: str, raise_exception: bool = False,
**kwargs) -> Optional[httpx.Response]:
"""
执行实际的异步请求
"""
kwargs.setdefault("headers", self._headers)
kwargs.setdefault("cookies", self._cookies)
try:
return await client.request(method, url, **kwargs)
except httpx.RequestError as e:
logger.debug(f"异步请求失败: {e}")
if raise_exception:
raise
return None
async def get(self, url: str, params: dict = None, **kwargs) -> Optional[str]:
"""
发送异步GET请求
:param url: 请求的URL
:param params: 请求的参数
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: 响应的内容，若发生RequestError则返回None
"""
response = await self.request(method="get", url=url, params=params, **kwargs)
if response:
try:
content = response.text
return content
except Exception as e:
logger.debug(f"处理异步响应内容失败: {e}")
return None
finally:
await response.aclose() # 确保连接被关闭
return None
async def post(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[httpx.Response]:
"""
发送异步POST请求
:param url: 请求的URL
:param data: 请求的数据
:param json: 请求的JSON数据
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
"""
return await self.request(method="post", url=url, data=data, json=json, **kwargs)
async def put(self, url: str, data: Any = None, **kwargs) -> Optional[httpx.Response]:
"""
发送异步PUT请求
:param url: 请求的URL
:param data: 请求的数据
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
"""
return await self.request(method="put", url=url, data=data, **kwargs)
async def get_res(self,
url: str,
params: dict = None,
data: Any = None,
json: dict = None,
allow_redirects: bool = True,
raise_exception: bool = False,
**kwargs) -> Optional[httpx.Response]:
"""
发送异步GET请求并返回响应对象
:param url: 请求的URL
:param params: 请求的参数
:param data: 请求的数据
:param json: 请求的JSON数据
:param allow_redirects: 是否允许重定向
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exception为True时会抛出
"""
return await self.request(method="get",
url=url,
params=params,
data=data,
json=json,
follow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
@asynccontextmanager
async def get_stream(self, url: str, params: dict = None, **kwargs):
"""
获取异步流式响应的上下文管理器，适用于大文件下载
:param url: 请求的URL
:param params: 请求的参数
:param kwargs: 其他请求参数
"""
kwargs['stream'] = True
response = await self.request(method="get", url=url, params=params, **kwargs)
try:
yield response
finally:
if response:
await response.aclose()
async def post_res(self,
url: str,
data: Any = None,
params: dict = None,
allow_redirects: bool = True,
files: Any = None,
json: dict = None,
raise_exception: bool = False,
**kwargs) -> Optional[httpx.Response]:
"""
发送异步POST请求并返回响应对象
:param url: 请求的URL
:param data: 请求的数据
:param params: 请求的参数
:param allow_redirects: 是否允许重定向
:param files: 请求的文件
:param json: 请求的JSON数据
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exception为True时会抛出
"""
return await self.request(method="post",
url=url,
data=data,
params=params,
follow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
async def put_res(self,
url: str,
data: Any = None,
params: dict = None,
allow_redirects: bool = True,
files: Any = None,
json: dict = None,
raise_exception: bool = False,
**kwargs) -> Optional[httpx.Response]:
"""
发送异步PUT请求并返回响应对象
:param url: 请求的URL
:param data: 请求的数据
:param params: 请求的参数
:param allow_redirects: 是否允许重定向
:param files: 请求的文件
:param json: 请求的JSON数据
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exception为True时会抛出
"""
return await self.request(method="put",
url=url,
data=data,
params=params,
follow_redirects=allow_redirects,
files=files,
json=json,
raise_exception=raise_exception,
**kwargs)
async def delete_res(self,
url: str,
data: Any = None,
params: dict = None,
allow_redirects: bool = True,
raise_exception: bool = False,
**kwargs) -> Optional[httpx.Response]:
"""
发送异步DELETE请求并返回响应对象
:param url: 请求的URL
:param data: 请求的数据
:param params: 请求的参数
:param allow_redirects: 是否允许重定向
:param raise_exception: 是否在发生异常时抛出异常，否则默认拦截异常返回None
:param kwargs: 其他请求参数，如headers, cookies, proxies等
:return: HTTP响应对象，若发生RequestError则返回None
:raises: httpx.RequestError 仅raise_exception为True时会抛出
"""
return await self.request(method="delete",
url=url,
data=data,
params=params,
follow_redirects=allow_redirects,
raise_exception=raise_exception,
**kwargs)
async def get_json(self, url: str, params: dict = None, **kwargs) -> Optional[dict]:
"""
发送异步GET请求并返回JSON数据，自动关闭连接
:param url: 请求的URL
:param params: 请求的参数
:param kwargs: 其他请求参数
:return: JSON数据，若发生异常则返回None
"""
response = self.request(method="get", url=url, params=params, **kwargs)
response = await self.request(method="get", url=url, params=params, **kwargs)
if response:
try:
data = response.json()
return data
except Exception as e:
logger.debug(f"解析JSON失败: {e}")
logger.debug(f"解析异步JSON失败: {e}")
return None
finally:
response.close()
await response.aclose()
return None
def post_json(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[dict]:
async def post_json(self, url: str, data: Any = None, json: dict = None, **kwargs) -> Optional[dict]:
"""
发送POST请求并返回JSON数据，自动关闭连接
发送异步POST请求并返回JSON数据，自动关闭连接
:param url: 请求的URL
:param data: 请求的数据
:param json: 请求的JSON数据
@@ -605,14 +853,14 @@ class RequestUtils:
"""
if json is None:
json = {}
response = self.request(method="post", url=url, data=data, json=json, **kwargs)
response = await self.request(method="post", url=url, data=data, json=json, **kwargs)
if response:
try:
data = response.json()
return data
except Exception as e:
logger.debug(f"解析JSON失败: {e}")
logger.debug(f"解析异步JSON失败: {e}")
return None
finally:
response.close()
await response.aclose()
return None

1
requirements.in
View File

@@ -71,3 +71,4 @@ setuptools~=78.1.0
pympler~=1.1
smbprotocol~=1.15.0
setproctitle~=1.3.6
httpx~=0.28.1