mirror of
https://github.com/xuthus83/LittlePaimon.git
synced 2024-12-16 13:40:53 +08:00
89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
import json
|
||
from pathlib import Path
|
||
from typing import Dict, Optional, Any, Union
|
||
|
||
import httpx
|
||
|
||
|
||
async def get(url: str,
|
||
*,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
timeout: Optional[int] = 20,
|
||
**kwargs) -> httpx.Response:
|
||
"""
|
||
说明：
|
||
httpx的get请求封装
|
||
参数：
|
||
:param url: url
|
||
:param headers: 请求头
|
||
:param params: params
|
||
:param timeout: 超时时间
|
||
"""
|
||
async with httpx.AsyncClient() as client:
|
||
return await client.get(url,
|
||
headers=headers,
|
||
params=params,
|
||
timeout=timeout,
|
||
**kwargs)
|
||
|
||
|
||
async def post(url: str,
|
||
*,
|
||
headers: Optional[Dict[str, str]] = None,
|
||
params: Optional[Dict[str, Any]] = None,
|
||
data: Optional[Dict[str, Any]] = None,
|
||
json: Optional[Dict[str, Union[Any, str]]] = None,
|
||
timeout: Optional[int] = 20,
|
||
**kwargs) -> httpx.Response:
|
||
"""
|
||
说明：
|
||
httpx的post请求封装
|
||
参数：
|
||
:param url: url
|
||
:param headers: 请求头
|
||
:param params: params
|
||
:param data: data
|
||
:param json: json
|
||
:param timeout: 超时时间
|
||
"""
|
||
async with httpx.AsyncClient() as client:
|
||
return await client.post(url,
|
||
headers=headers,
|
||
params=params,
|
||
data=data,
|
||
json=json,
|
||
timeout=timeout,
|
||
**kwargs)
|
||
|
||
|
||
def load_json(path: Union[Path, str], encoding: str = 'utf-8') -> dict:
|
||
"""
|
||
说明：
|
||
读取本地json文件，返回json字典。
|
||
参数：
|
||
:param path: 文件路径
|
||
:param encoding: 编码，默认为utf-8
|
||
:return: json字典
|
||
"""
|
||
if isinstance(path, str):
|
||
path = Path(path)
|
||
if not path.exists():
|
||
save_json({}, path, encoding)
|
||
return json.load(path.open('r', encoding=encoding))
|
||
|
||
|
||
def save_json(data, path: Union[Path, str], encoding: str = 'utf-8'):
|
||
"""
|
||
说明：
|
||
将数据保存至本地json文件
|
||
参数：
|
||
:param data: json数据
|
||
:param path: 文件路径
|
||
:param encoding: 编码，默认为utf-8
|
||
"""
|
||
if isinstance(path, str):
|
||
path = Path(path)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
json.dump(data, path.open('w', encoding=encoding), ensure_ascii=False, indent=4)
|