|
# retoor <retoor@molodetz.nl>
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
from fastapi import Request
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BODY_METHODS = {"POST", "PUT", "DELETE", "PATCH"}
|
|
|
|
|
|
async def merge_params(request: Request) -> dict:
|
|
params: dict[str, Any] = dict(request.query_params)
|
|
if request.method not in BODY_METHODS:
|
|
return params
|
|
content_type = request.headers.get("content-type", "")
|
|
if "application/json" in content_type:
|
|
try:
|
|
body = await request.json()
|
|
except (json.JSONDecodeError, UnicodeDecodeError, ValueError):
|
|
body = None
|
|
if isinstance(body, dict):
|
|
params.update(body)
|
|
return params
|
|
try:
|
|
form = await request.form()
|
|
except (ValueError, RuntimeError) as error:
|
|
logger.debug("devrant form parse failed: %s", error)
|
|
return params
|
|
for key in form:
|
|
params[key] = form[key]
|
|
return params
|