Skip to content

Commit

Permalink
Lazy compute all routes from FastAPI app once
Browse files Browse the repository at this point in the history
Instead of keeping a reference to the FastAPI app, I think it's better
to keep its routes. This way, on `HALFor()`, it's more convenient to
retrieve a given route by its endpoint (aka route.name) rather than
iterate on app.routes at every call.
  • Loading branch information
wo0dyn committed Mar 16, 2023
1 parent c1ec756 commit c46ce44
Show file tree
Hide file tree
Showing 6 changed files with 72 additions and 44 deletions.
2 changes: 1 addition & 1 deletion docs/basics.md
Expand Up @@ -83,7 +83,7 @@ Alternatively, rather than providing the endpoint name, you can provide a refere

### `values` (optional depending on endpoint)

Same keyword arguments as FastAPI's url_path_for, except string arguments enclosed in < > will be interpreted as attributes to pull from the object. For example, here we need to pass an `item_id` argument as required by our endpoint function, and we want to populate that with our item object's `id` attribute.
Same keyword arguments as FastAPI's `url_path_for()`, except string arguments enclosed in < > will be interpreted as attributes to pull from the object. For example, here we need to pass an `item_id` argument as required by our endpoint function, and we want to populate that with our item object's `id` attribute.

## Create a link set

Expand Down
13 changes: 6 additions & 7 deletions docs/extending.md
Expand Up @@ -64,7 +64,7 @@ class UrlFor(UrlType):
At this point, our custom type will behave as a normal Pydantic type, but won't do any hypermedia substitutions.
For this, we must add our "magic" `__build_hypermedia__` method.

```python hl_lines="32-38"
```python hl_lines="32-37"
from fastapi_hypermodel.hypermodel import UrlType, resolve_param_values
from starlette.datastructures import URLPath

Expand Down Expand Up @@ -97,12 +97,11 @@ class UrlFor(UrlType, AbstractHyperField):
)

def __build_hypermedia__(
self, app: Optional[FastAPI], values: Dict[str, Any]
self, routes: Dict[str, Route], values: Dict[str, Any]
) -> Optional[str]:
if app is None:
return None
route = routes[self.endpoint]
resolved_params = resolve_param_values(self.param_values, values)
return app.url_path_for(self.endpoint, **resolved_params)
return route.url_path_for(self.endpoint, **resolved_params)
```

Here we see that, as expected, our method accepts a `FastAPI` instance, and our dict of parent field values. We pass these field values, along with the URL parameter to field mappings, to a `resolve_param_values` function. This function takes our URL parameter to field mappings, and substitutes in the *actual* values from the parent.
Expand All @@ -128,9 +127,9 @@ class LinkSet(_LinkSetType, AbstractHyperField): # pylint: disable=too-many-anc
field_schema.update({"additionalProperties": _uri_schema})

def __build_hypermedia__(
self, app: Optional[FastAPI], values: Dict[str, Any]
self, routes: Dict[str, Route], values: Dict[str, Any]
) -> Dict[str, str]:
return {k: u.__build_hypermedia__(app, values) for k, u in self.items()} # type: ignore # pylint: disable=no-member
return {k: u.__build_hypermedia__(routes, values) for k, u in self.items()} # type: ignore # pylint: disable=no-member
```

This class behaves link a standard dictionary, with `str` keys and any other `AbstractHyperField` as values. This allows, for example, nesting `LinkSet` instances for rich, deep hypermedia, as well as allowing different hypermedia types (such as `HALFor` links).
Expand Down
2 changes: 1 addition & 1 deletion examples/simple_app.py
Expand Up @@ -70,7 +70,7 @@ class Person(HyperModel):
"person02": {"id": "person02", "name": "Bob", "items": [items["item03"]]},
}

# Create our API routes, using our Pydantic models as respone_model
# Create our API routes, using our Pydantic models as response_model


@app.get("/items", response_model=List[ItemSummary])
Expand Down
77 changes: 51 additions & 26 deletions fastapi_hypermodel/hypermodel.py
Expand Up @@ -6,14 +6,24 @@
import abc
import re
import urllib
from typing import Any, Callable, Dict, List, Optional, Union, no_type_check
from typing import (
Any,
Callable,
Dict,
Iterable,
List,
Optional,
Tuple,
Union,
no_type_check,
)

from fastapi import FastAPI
from pydantic import BaseModel, PrivateAttr, root_validator
from pydantic.utils import update_not_none
from pydantic.validators import dict_validator
from starlette.datastructures import URLPath
from starlette.routing import Route
from starlette.routing import BaseRoute, Host, Mount, Route

_tpl_pattern = re.compile(r"\s*<\s*(\S*)\s*>\s*")

Expand All @@ -26,7 +36,7 @@ class InvalidAttribute(AttributeError):

class AbstractHyperField(metaclass=abc.ABCMeta):
@abc.abstractmethod
def __build_hypermedia__(self, app: Optional[FastAPI], values: Dict[str, Any]):
def __build_hypermedia__(self, routes: Dict[str, Route], values: Dict[str, Any]):
return


Expand Down Expand Up @@ -74,14 +84,17 @@ def validate(cls, value: Any) -> "UrlFor":
)

def __build_hypermedia__(
self, app: Optional[FastAPI], values: Dict[str, Any]
self, routes: Dict[str, Route], values: Dict[str, Any]
) -> Optional[str]:
if app is None:
return None
if self.condition is not None and not self.condition(values):
return None

route = routes.get(self.endpoint, None)
if route is None:
raise ValueError(f"No route found for endpoint {self.endpoint}")

resolved_params = resolve_param_values(self.param_values, values)
return app.url_path_for(self.endpoint, **resolved_params)
return route.url_path_for(self.endpoint, **resolved_params)


class HALItem(BaseModel):
Expand Down Expand Up @@ -110,29 +123,20 @@ def __init__(
super().__init__()

def __build_hypermedia__(
self, app: Optional[FastAPI], values: Dict[str, Any]
self, routes: Dict[str, Route], values: Dict[str, Any]
) -> Optional[HALItem]:
if app is None:
return None
if self._condition is not None and not self._condition(values):
return None

resolved_params = resolve_param_values(self._param_values, values)

this_route = next(
(
route
for route in app.routes
if isinstance(route, Route) and route.name == self._endpoint
),
None,
)
if not this_route:
route = routes.get(self._endpoint, None)
if route is None:
raise ValueError(f"No route found for endpoint {self._endpoint}")

return HALItem(
href=app.url_path_for(self._endpoint, **resolved_params),
method=next(iter(this_route.methods), None) if this_route.methods else None,
href=route.url_path_for(self._endpoint, **resolved_params),
method=next(iter(route.methods), None) if route.methods else None,
description=self._description,
)

Expand All @@ -150,9 +154,9 @@ def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:
field_schema.update({"additionalProperties": _uri_schema})

def __build_hypermedia__(
self, app: Optional[FastAPI], values: Dict[str, Any]
self, routes: Dict[str, Route], values: Dict[str, Any]
) -> Dict[str, str]:
links = {k: u.__build_hypermedia__(app, values) for k, u in self.items()} # type: ignore # pylint: disable=no-member
links = {k: u.__build_hypermedia__(routes, values) for k, u in self.items()} # type: ignore # pylint: disable=no-member
return {k: u for k, u in links.items() if u is not None}


Expand Down Expand Up @@ -231,14 +235,19 @@ def resolve_param_values(


class HyperModel(BaseModel):
_hypermodel_bound_app: Optional[FastAPI] = None
_hypermodel_bound_routes: Optional[Dict[str, Route]] = None

@root_validator
def _hypermodel_gen_href(cls, values): # pylint: disable=no-self-argument
if (
cls._hypermodel_bound_routes is None
and (app := getattr(cls, "__hypermodel_bound_app", None)) is not None
):
cls._hypermodel_bound_routes = dict(cls.get_all_routes(app.routes))
for key, value in values.items():
if isinstance(value, AbstractHyperField):
values[key] = value.__build_hypermedia__(
cls._hypermodel_bound_app, values
cls._hypermodel_bound_routes, values
)
return values

Expand All @@ -252,4 +261,20 @@ def init_app(cls, app: FastAPI):
Args:
app (FastAPI): Application to generate URLs from
"""
cls._hypermodel_bound_app = app
setattr(cls, "__hypermodel_bound_app", app)

@staticmethod
def get_all_routes(app_routes: List[BaseRoute]) -> Iterable[Tuple[str, Route]]:
"""
Helper to retrieve all routes from app.routes.
Args:
app_routes (List[BaseRoute]): List of routes from a FastAPI app.
"""
for route in app_routes:
if isinstance(route, (Host, Mount)):
# Host and Mount have a `routes` property, so we have
# to iterate on each one of them recursively.
yield from HyperModel.get_all_routes(route.routes)
elif isinstance(route, Route):
yield route.name, route
8 changes: 2 additions & 6 deletions tests/app.py
Expand Up @@ -3,8 +3,7 @@
from fastapi import FastAPI
from pydantic.main import BaseModel

from fastapi_hypermodel import HyperModel, LinkSet, UrlFor
from fastapi_hypermodel.hypermodel import HALFor
from fastapi_hypermodel import HALFor, HyperModel, LinkSet, UrlFor


class ItemSummary(HyperModel):
Expand Down Expand Up @@ -122,10 +121,7 @@ def update_item(item_id: str, item: ItemUpdate):
return items[item_id]


@test_app.get(
"/people",
response_model=List[Person],
)
@test_app.get("/people", response_model=List[Person])
def read_people():
return list(people.values())

Expand Down
14 changes: 11 additions & 3 deletions tests/test_fastapi_hypermodel.py
Expand Up @@ -159,16 +159,24 @@ def test_people_halset_condition_unmet(client, person_id):
pytest.param(read_item, id="Use of a Callable endpoint"),
],
)
def test_bad_attribute(app, endpoint):
def test_bad_attribute(endpoint):
class ItemSummary(HyperModel):
href = UrlFor(endpoint, {"item_id": "<id>"})

assert ItemSummary._hypermodel_bound_app is app

with pytest.raises(InvalidAttribute):
_ = ItemSummary()


def test_app_routes(app):
"""
Check that all routes defined in the FastAPI app are in the HyperModel object.
"""
assert HyperModel() # Force computing routes
assert {route.name for route in app.routes} == set(
HyperModel._hypermodel_bound_routes
)


### APP TESTS, SHOULD REMOVE


Expand Down

0 comments on commit c46ce44

Please sign in to comment.