49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
import httpx
|
|
import logging
|
|
from pydantic import BaseModel
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def docstring_from_model(model: type[BaseModel]):
|
|
def decorator(func):
|
|
if func.__doc__:
|
|
func.__doc__ = func.__doc__.format(
|
|
params="\n".join(
|
|
f"\t\t- {name}: {field.description or 'No description'}"
|
|
for name, field in model.model_fields.items()
|
|
)
|
|
)
|
|
return func
|
|
return decorator
|
|
|
|
async def fetch_api_data(icon: str, url: str, params: dict, remove_keys: list = None) -> dict:
|
|
try:
|
|
logger.info(f"\n🔨{icon}Input parameters: {params}\n")
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(url, params=params, timeout=10.0)
|
|
response.raise_for_status()
|
|
logger.debug(f"\n🖇️{icon}Request URL: {response.url}\n")
|
|
|
|
data = response.json()
|
|
if remove_keys and isinstance(data, dict):
|
|
for key in remove_keys:
|
|
data.pop(key, None)
|
|
logger.info(f"\n🚮{icon}Successfully removed key: {key}\n")
|
|
|
|
logger.info(f"\n✅{icon}Successfully fetched data from {url}\n")
|
|
return data
|
|
|
|
except httpx.HTTPStatusError as e:
|
|
logger.error(f"\n{icon}HTTP error: {e.response.status_code} - {e.response.text}\n")
|
|
return {"error": "http_error", "status_code": e.response.status_code, "detail": e.response.text}
|
|
except httpx.RequestError as e:
|
|
logger.error(f"\n{icon}Request error: {str(e)}\n")
|
|
return {"error": "request_error", "status_code": str(e)}
|
|
except Exception as e:
|
|
logger.critical(f"\n{icon}Unexpected error: {str(e)}\n", exc_info=True)
|
|
return {"error": "unexpected_error", "status_code": str(e)}
|