mirror of
https://gitlab.archlinux.org/archlinux/aurweb.git
synced 2025-02-03 10:43:03 +01:00
6662975005
Additionally, added RPC.error, which produces an RPC-compatible error based on the version passed during construction. Signed-off-by: Kevin Morris <kevr@0cost.org>
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
from typing import List, Optional
|
|
from urllib.parse import unquote
|
|
|
|
from fastapi import APIRouter, Query, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
from aurweb.rpc import RPC
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
def parse_args(request: Request):
|
|
""" Handle legacy logic of 'arg' and 'arg[]' query parameter handling.
|
|
|
|
When 'arg' appears as the last argument given to the query string,
|
|
that argument is used by itself as one single argument, regardless
|
|
of any more 'arg' or 'arg[]' parameters supplied before it.
|
|
|
|
When 'arg[]' appears as the last argument given to the query string,
|
|
we iterate from last to first and build a list of arguments until
|
|
we hit an 'arg'.
|
|
|
|
TODO: This handling should be addressed in v6 of the RPC API. This
|
|
was most likely a bi-product of legacy handling of versions 1-4
|
|
which we no longer support.
|
|
|
|
:param request: FastAPI request
|
|
:returns: List of deduced arguments
|
|
"""
|
|
# Create a list of (key, value) pairs of the given 'arg' and 'arg[]'
|
|
# query parameters from last to first.
|
|
query = list(reversed(unquote(request.url.query).split("&")))
|
|
parts = [
|
|
e.split("=", 1) for e in query if e.startswith(("arg=", "arg[]="))
|
|
]
|
|
|
|
args = []
|
|
if parts:
|
|
# If we found 'arg' and/or 'arg[]' arguments, we begin processing
|
|
# the set of arguments depending on the last key found.
|
|
last = parts[0][0]
|
|
|
|
if last == "arg":
|
|
# If the last key was 'arg', then it is our sole argument.
|
|
args.append(parts[0][1])
|
|
else:
|
|
# Otherwise, it must be 'arg[]', so traverse backward
|
|
# until we reach a non-'arg[]' key.
|
|
for key, value in parts:
|
|
if key != last:
|
|
break
|
|
args.append(value)
|
|
|
|
return args
|
|
|
|
|
|
@router.get("/rpc")
|
|
async def rpc(request: Request,
|
|
v: Optional[int] = Query(None),
|
|
type: Optional[str] = Query(None),
|
|
arg: Optional[str] = Query(None),
|
|
args: Optional[List[str]] = Query(None, alias="arg[]")):
|
|
|
|
# Create a handle to our RPC class.
|
|
rpc = RPC(version=v, type=type)
|
|
|
|
# Prepare list of arguments for input. If 'arg' was given, it'll
|
|
# be a list with one element.
|
|
arguments = parse_args(request)
|
|
return JSONResponse(rpc.handle(arguments))
|