Spaces:
Running
Running
File size: 3,433 Bytes
f924769 780e0d1 f924769 259a596 f924769 780e0d1 f924769 259a596 9316f10 259a596 f924769 780e0d1 f924769 780e0d1 f924769 6a8befc f924769 |
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 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 |
from fastapi import Query
from typing import Dict, Literal
import httpx
import os
def extract_type(p):
schema = p.get("schema", {})
if "$ref" in schema:
return str # fallback
if "enum" in schema:
return Literal[tuple(schema["enum"])]
return openapi_type_to_python(schema.get("type", "string"))
def get_param_metadata(p):
schema = p.get("schema", {})
return {
"name": p["name"],
"type": extract_type(p),
"required": p.get("required", False),
"description": p.get("description", ""),
"default": schema.get("default", ...), # `...` => required if undefined
"enum": schema.get("enum", None),
}
async def get_remote_params(base_url: str,
endpoint: str,
method: str = 'get') -> Dict:
async with httpx.AsyncClient() as client:
headers = {}
if TENNIS_ML_API_KEY := os.getenv('TENNIS_ML_API_KEY'):
headers["Authorization"] = TENNIS_ML_API_KEY
response = await client.get(f"{base_url}openapi.json", headers=headers)
spec = response.json()
# Extraction of the parameters from the OpenAPI spec
path_def = spec["paths"].get(f'/{endpoint}', {})
method_def = path_def.get(method, {})
params = method_def.get("parameters", [])
return {
"found": bool(path_def),
"general": {
"description": method_def.get("description", ""),
"summary": method_def.get("summary", ""),
"tags": method_def.get("tags", []),
},
"params": [
get_param_metadata(p)
for p in params if p["in"] == "query"
]
}
def openapi_type_to_python(t: str):
return {
"string": str,
"integer": int,
"number": float,
"boolean": bool,
"array": list,
}.get(t, str) # fallback = str
def create_forward_endpoint(base_url: str, _endpoint: str, param_defs: Dict):
# Dynamic endpoint creation
def endpoint_factory():
async def endpoint(**kwargs):
headers = {}
if TENNIS_ML_API_KEY := os.getenv('TENNIS_ML_API_KEY'):
headers["Authorization"] = TENNIS_ML_API_KEY
async with httpx.AsyncClient() as client:
response = await client.get(
f"{base_url}{_endpoint}",
params=kwargs,
headers=headers,
)
response.raise_for_status()
return response.json()
return endpoint
endpoint = endpoint_factory()
# Add the parameters to the endpoint
endpoint.__annotations__ = {
p["name"]: openapi_type_to_python(p["type"]) for p in param_defs
}
# Add dependencies to the endpoint
from inspect import Parameter, Signature
parameters = []
for p in param_defs:
annotation = p["type"]
default = p.get("default", ...) if p.get("required", False) else p.get("default", None)
query = Query(
default=default,
description=p.get("description", "")
)
param = Parameter(
name=p["name"],
kind=Parameter.KEYWORD_ONLY,
default=query,
annotation=annotation
)
parameters.append(param)
endpoint.__signature__ = Signature(parameters)
return endpoint
|