Created
September 3, 2020 12:12
-
-
Save jacksmith15/b220686eae16d52a10ef59b5de09c213 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
"""A set of helpers for reversing urls, similar to Django ``reverse``. | |
Usage: | |
.. code:: python | |
@router.get("/class/{class_id}") | |
async def get_class(request: Request, class_id: int = Path(...)): | |
student_route = get_route(request.app, "list_students") | |
class_students_url = URLFactory(student_route).get_path(class_id=class_id) | |
return { | |
"id": class_id, | |
"students": class_students_url | |
} | |
Usage can be simplified by binding to a subclass of ``fastapi.Request`` or | |
``fastapi.FastAPI``. Example: | |
.. code:: python | |
class MyRequest(Request): | |
def reverse(self, name: str, **kwargs) -> str: | |
return URLFactory(get_route(self.app, name)).get_url( | |
SETTINGS.BASE_URL, **kwargs | |
) | |
@router.get("/class/{class_id}") | |
async def get_class(request: MyRequest, class_id: int = Path(...)): | |
return { | |
"id": class_id, | |
"students": request.reverse("list_students", class_id=class_id) | |
} | |
""" | |
import json | |
from typing import cast | |
from urllib.parse import unquote | |
from fastapi.routing import APIRoute | |
from fastapi import FastAPI | |
from fastapi.dependencies.utils import request_params_to_args | |
from furl import furl | |
from pydantic import BaseModel, create_model, ValidationError | |
def get_route(app: FastAPI, name: str) -> APIRoute: | |
"""Get a route by name from a ``FastAPI`` application.""" | |
results = [ | |
route | |
for route in app.routes | |
if isinstance(route, APIRoute) | |
and route.name == name | |
and route.methods | |
and "GET" in route.methods | |
] | |
if not results: | |
raise KeyError(f"No GET route registered with name: {name}") | |
return results[0] | |
URLErrorModel = create_model("URLErrorModel") | |
"""Dummy model with which to bind validation errors on URL resolution.""" | |
class ParamModel(BaseModel): | |
"""Model for rendering parameters as ``json``-friendly python types.""" | |
class Config: | |
extra = "allow" | |
def render(self): | |
return json.loads(self.json()) | |
class URLFactory: | |
"""Factory for generating valid URLs from ``fastapi.APIRoute``. | |
Leverages the same parameter validation and error reporting used at | |
runtime. Only supports path and query parameters, as other parameters | |
are not expressed in the URL. | |
""" | |
def __init__(self, route: APIRoute): | |
self.dependant = route.dependant | |
self.path_param_names = {param.alias for param in self.dependant.path_params} | |
self.query_param_names = {param.alias for param in self.dependant.query_params} | |
def get_path(self, **kwargs) -> str: | |
"""Resolve a path for the ``APIRoute``, validating path and query parameters. | |
:param kwargs: Path and query parameters to apply to the url. | |
""" | |
request = kwargs.pop("request") | |
if request: | |
kwargs = {**request.query_params, **kwargs} | |
path_params = ParamModel( | |
**{key: value for key, value in kwargs.items() if key in self.path_param_names} | |
).render() | |
query_params = ParamModel( | |
**{key: value for key, value in kwargs.items() if key in self.query_param_names} | |
).render() | |
path_values, path_errors = request_params_to_args( | |
self.dependant.path_params, path_params | |
) | |
query_values, query_errors = request_params_to_args( | |
self.dependant.query_params, query_params | |
) | |
errors = path_errors + query_errors | |
if errors: | |
raise ValidationError(errors, URLErrorModel) | |
path = furl(cast(str, self.dependant.path).format(**path_params)) | |
for key, value in query_params.items(): | |
path.args[key] = value | |
return unquote(str(path)) | |
def get_url(self, base_url: str, **kwargs) -> str: | |
"""Resolve a complete URL for the ``APIRoute``, validating path and query parameters. | |
:param base_url: The base url of the API. | |
:param kwargs: Path and query parameters to apply to the url. | |
""" | |
return base_url.rstrip("/") + "/" + self.get_path(**kwargs).lstrip("/") |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment