Spaces:
Sleeping
Sleeping
File size: 2,169 Bytes
1b7e88c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 |
import httpx
import requests
from .error import VQLError
from .logger import logging
def request(
json: dict,
url: str,
headers: dict | None = None,
) -> dict:
try:
res = requests.post(url=url, json=json, headers=headers)
except Exception as error:
logging.error(
"Request URL: {} | Request body: {} | Error: {}".format(url, json, error)
)
raise VQLError(511, detail=str(error))
if res.status_code < 299:
return res.json()
elif res.status_code == 404:
logging.error("Request URL: {} | Error[404]: 请求错误: 错误的地址".format(url))
raise VQLError(516)
elif res.status_code == 422:
logging.error(
"Request URL: {} | Request body: {} | Error[422]: 请求错误: 错误的请求格式".format(
url, json
)
)
raise VQLError(517)
else:
info = res.json()
logging.error(
"Request URL: {} | Request body: {} | Error: {}".format(url, json, info)
)
raise VQLError(511, detail=info)
async def arequest(url: str, json: dict, headers: dict | None = None) -> dict:
try:
async with httpx.AsyncClient() as client:
res = await client.post(url=url, json=json, headers=headers, timeout=30)
except Exception as error:
logging.error(
"Request URL: {} | Request body: {} | Error: {}".format(url, json, error)
)
raise VQLError(511, detail=str(error))
if res.status_code < 299:
return res.json()
elif res.status_code == 404:
logging.error("Request URL: {} | Error[404]: 请求错误: 错误的地址".format(url))
raise VQLError(516)
elif res.status_code == 422:
logging.error(
"Request URL: {} | Request body: {} | Error[422]: 请求错误: 错误的请求格式".format(
url, json
)
)
raise VQLError(517)
else:
info = res.json()
logging.error(
"Request URL: {} | Request body: {} | Error: {}".format(url, json, info)
)
raise VQLError(511, detail=info)
|