-
First check
PreamblePlease add support for class-based encapsulated route definitions. This is similar to fastapiutils There are many motivations for this, but the biggest to me are:
This is achievable using global variables or re-importing files, and maybe that is the pythonic way, but that always becomes problematic in large projects. I would really love to use FastAPI at my company but can't quite justify it given the lack of encapsulation. Django makes this possible, but I prefer not to use Django because of all the legacy support baggage they carry. Testing FastAPI projects without this requires global definitions of shared resources and there is no nice way run dependency injection per environment etc without attaching a global config (again, maybe this is the pythonic way and I need to stop thinking in Java terms... lmk). Wouldn't it be great if we could define a router like this: from fastapi import FastAPI
from fastapi.routing import LateBoundAPIRouter
from fastapi.testclient import TestClient
from pydantic import BaseModel
router = LateBoundAPIRouter(instance_delegate=lambda: EncapsulatedRoutes.INSTANCE)
app = FastAPI()
class Item(BaseModel):
message_update: str
class EncapsulatedRoutes:
INSTANCE = None
def __init__(self, message: str):
EncapsulatedRoutes.INSTANCE = self
self.message = message
@router.get("/message")
def get_message(self) -> str:
return self.message
@router.post("/message")
def post_message(self, item: Item) -> str:
self.message = item.message_update
app.include_router(router, is_late_bound=True)
client = TestClient(app)
instance = EncapsulatedRoutes(message="👋")
def test_late_bound_router():
response = client.get("/message")
assert response.status_code == 200, response.text
assert instance.message in response.text
item = Item(message_update="✨")
response = client.post("/message", json=item.dict())
assert response.status_code == 200, response.text
response = client.get("/message")
assert response.status_code == 200, response.text
assert item.message_update in response.textDescriptionFrom this you can easily move the class to a separate file without polluting the global namespace. Tests can inject mocked versions of stateful dependencies. Stateful routes can manage their state without the risk of putting internal objects into the global namespace where they might be modified by other classes (stranger things have happened on large projects). None of this is 100% required, but it would greatly enhance the testability and modularity of FastAPI. Describe alternatives you've consideredThere is some precedent here with FastAPI Utils, and the class-based-view It is not very future-proof and is likely to develop or already contain hard-to-find bugs. Environment
Additional contextAt its core this is a 'chicken and egg' style problem as the router decorators must be defined before they can be attached to the class. Some reflective magic could be added to auto-wire this based on class name etc but that breaks the the pythonic astonishment principle significantly. The simplest and most pythonic way that I see to make this possible is to add a hook in routing.py where a flag or some other indicator causes the So basically two changes needed:
class LateBoundAPIRouter(APIRouter):
def __init__(
self, instance_delegate: Callable[..., Any], *args: Any, **kwargs: Any
):
super(LateBoundAPIRouter, self).__init__(*args, **kwargs)
self.instance_delegate = instance_delegate
def add_api_route(
self, path: str, endpoint: Callable[..., Any], **kwargs: Any
) -> None:
@functools.wraps(endpoint)
def endpoint_wrapper(*args: Any, **kwargs: Any) -> Any:
return endpoint(self.instance_delegate(), *args, **kwargs)
super().add_api_route(path, endpoint_wrapper, is_late_bound=True, **kwargs)There are a few ways to improve robustness here, but good docs and possibly singleton enforcement could solve that. Some other options I see are:
If there is appetite for this among the maintainers I would be happy to implement it. |
Beta Was this translation helpful? Give feedback.
Replies: 10 comments 3 replies
-
|
Can someone elaborate on the 👎 s? I see a lot of requests for this feature and this is a lot better than trying to patch it in from a utils package. |
Beta Was this translation helpful? Give feedback.
-
|
I found a pretty nice way to do this - minimal impact to public API surface and error handling for edge cases included. It overloads the view initializer to apply route binding during view initialization. from fastapi import FastAPI
from fastapi.routing import ViewAPIRouter
from fastapi.testclient import TestClient
from pydantic import BaseModel
view_router = ViewAPIRouter()
app = FastAPI()
class Item(BaseModel):
message_update: str
@view_router.bind_to_class()
class Messages(ViewAPIRouter.View):
def __init__(self, message: str):
self.message = message
@view_router.get("/message")
def get_message(self) -> str:
return self.message
@view_router.post("/message")
def post_message(self, item: Item) -> str:
self.message = item.message_update
em_instance = Messages(message="👋")
pt_instance = Messages(message="olá")
en_instance = Messages(message="hello")
app.include_router(em_instance.router, prefix="/em")
app.include_router(pt_instance.router, prefix="/pt")
app.include_router(en_instance.router, prefix="/en")
client = TestClient(app)
# verify route inclusion
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert em_instance.message in response.text
response = client.get("/pt/message")
assert response.status_code == 200, response.text
assert pt_instance.message in response.text
response = client.get("/en/message")
assert response.status_code == 200, response.text
assert en_instance.message in response.text
# change state in an instance
item = Item(message_update="✨")
response = client.post("/em/message", json=item.dict())
assert response.status_code == 200, response.text
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert item.message_update in response.text |
Beta Was this translation helpful? Give feedback.
-
|
This is an elegant implementation. However it appears that ViewAPIRouter() no longer exists? How would one do this with the current version of FastAPI? |
Beta Was this translation helpful? Give feedback.
-
|
Hey @Kojiro20, thanks for your interest. I see that your approach is different from About your motivations:
I think you can achieve more or less the same functionality you are showing here with a function, without needing any additional complexity or changes in FastAPI: from fastapi import FastAPI, APIRouter
from fastapi.testclient import TestClient
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
message_update: str
def get_customized_router(message: str):
router = APIRouter()
@router.get("/message")
def get_message() -> str:
return message
@router.post("/message")
def post_message(item: Item) -> str:
# Not sure what would be the intention here, probably save in DB?
# self.message = item.message_update
return f"Done with {item.message_update}"
em_router = get_customized_router(message="👋")
pt_router = get_customized_router(message="olá")
en_router = get_customized_router(message="hello")
app.include_router(em_router, prefix="/em")
app.include_router(pt_router, prefix="/pt")
app.include_router(en_router, prefix="/en")
client = TestClient(app)
def test_all():
# verify route inclusion
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert "👋" in response.text
response = client.get("/pt/message")
assert response.status_code == 200, response.text
assert "olá" in response.text
response = client.get("/en/message")
assert response.status_code == 200, response.text
assert "hello" in response.text
# change state in an instance
item = Item(message_update="✨")
response = client.post("/em/message", json=item.dict())
assert response.status_code == 200, response.text
response = client.get("/em/message")
assert response.status_code == 200, response.text
assert item.message_update in response.text
You probably shouldn't allocate and use a database connection per class/instance/resource. It would probably be better to have a connection per application, or even better, a shared connection pool so that they are all efficiently shared/used by all your path operations through all the requests. You might want to create a session per request. But it would normally be per request for a single client, not per class handling similar data that is not used in the same request. I'm not sure what you mean with "stateful tasks". If you refer to the section: self.message = item.message_update...then it would probably be better to do it in a different way. Otherwise, it would mean that your application would lose all its state after a restart, and if you have more than one process (that you most probably will in production), they would all have a different state. And if you have more than one concurrent client you will probably end up in race conditions with clients overriding the data for others.
I'm not sure what you mean by that. You can probably use the included dependency injection system, and the ideas to test them.
Can you elaborate on that? Why do you think that approach is so much worse than the proposed one? |
Beta Was this translation helpful? Give feedback.
-
|
Thank you for the reply. I really appreciate it. I think root cause may be that I don't like python as a language 🙃 . So, I need to get over that. I understand that functionally classes and functions are quite similar but the object oriented approach feels nicer to me - even if it's more of an articulation of intent than an enforceable contract as it is with strongly typed languages. The function approach is similar to what I've been doing. Something like: class ARedisBackedRoute(SomeBaseMaybeWithRedisToolsForTTLAndCommonHighHitRateEntities):
def __init__(self, extra_stuff, *args, **kwargs):
# please let me know if there's a better way to manage inheritance
super(ARedisBackedRoute._Instance, self).__init__(*args, **kwargs)
# note: at this point self.router has been initialized and partially composed
self.extra_stuff = extra_stuff
self.build_routes()
def build_routes(self):
@self.router.get("/message")
def get_message() -> str:
return message
@self.router.post("/message")
def post_message(item: Item) -> str:
# Not sure what would be the intention here, probably save in DB?
# self.message = item.message_update
return f"Done with {item.message_update}"
return routerI don't like this because the actual route definitions are unreachable. If I ever encountered a situation where using TestClient became more of an 'integration test' than a 'unit test' I would want a way to invoke the method directly instead of standing up an app container. I can also abstract the implementations of methods from their declarations - kind of two-layers for def build_routes(self):
@self.router.get("/message")
def get_message() -> str:
return self.get_message()
def get_message(self):
return 'message'
This is true - but I think it becomes more interesting when you surpass the load capacity for one database instance. For example, if a sharding strategy, such as consistent hashing is needed. Some clients will automagically provide support for this through their client interface, but it is precarious to depend on pre-configured logic when your service reaches such a scale inflection point. In some cases the sharding decision can depend on customer/client/user parameters that aren't available or shouldn't be known by the database connection pool. Separate deployment stamps might also have separate strategies for tiered customers etc. It is possible to do this through a single global instance import. But, it starts to bleed implementation details of specific components throughout the app. For example, if I wanted to do this with a global db connection instance I would probably need to wrap it and throw in extra kwargs to it's accessor and add component-specific logic that would be applied to all consumers but only applicable to one use-case.
Good point, this is a bad example. Something like a first-tier defensive rate limiter might be better. Actually, a session cache is a good fit here. If my service is proxying client interactions to something like a STUN/TURN server I might implement an LRU session cache so that blips in client connectivity don't always force renegotiation of backend connections (i.e. if the client is lucky enough to hit a handler in same process - or if their connectivity is really bad the session will eventually be cached in all process instances). Another example would be cached state for long-polling. If I want to hold the connection until a value has changed or until some time has elapsed. I needed a way to manage that state in the route handler (For what it's worth, this wasn't pretty. But, it was the least-worst way to reduce load from legacy clients that polled the service every second for changes - using this we were able to hold the connection for ~30s if nothing changed, but still respond immediately when the requested resource had an update).
Thank you, I didn't realize this was a general dependency injection system. I'll try this. At first glance it looks like it still relies on global instance imports. Which, I am not a fan of - but perhaps I need to get over that.
At a high-level my concern is that cbvs in fastapi_utils rewrite the function signature, essentially breaking intent with language internals. I think the fact that this is possible (even on things that the language authors tried to make immutable) is a good summary for why I have issues with python as a language in general. If I understand the new charter/intent of current project owners for the python language (since Guido left), they want to stop things like this from being possible. There is a gradual progression toward true type safety. I have doubts that they will be able to do this, but it is a shaky foundation for the feature. When I have played with things like this in the past (for example to implement long-polling based on changes to an object) I often learn something new about the language and discover more reasons why messing with internals is perilous. I like my approach better because it respects language internals and resolves dropping of the Again - thank you for responding here, I know you're busy. Feel free to close this and my PR if it's not aligned to the project goals. |
Beta Was this translation helpful? Give feedback.
-
|
I was looking for a way to use a class as a Controller (or Router) to be able to inject in one place (constructor) the dependencies. Otherwise, it feels we do Dry Code with that dependencies. It would be so much pleasure to have this possibility to deal with the router of fast API. Don't you think this kind of PR could bring a lot of good to the framework? Especially when you have DI ready. |
Beta Was this translation helpful? Give feedback.
-
|
I'm thinking along similar lines, not only wanting to inject dependencies in one place (although just that would be nice) but also wanting to be able to write a base Route class for e.g. CRUD operations that can be inherited and augmented as needed. But back to the dependency injection, would it be a big change to fastAPI if instead of allowing just a list of dependencies to be defined on the APP / Router instance to also allow dictionaries of dependencies? and then expose that dictionary in any function that has the APP / Router as a decorator? I haven't looked at the code of these dependency injections but happy to collaborate on a PR that would take a stab at it. |
Beta Was this translation helpful? Give feedback.
-
|
Thanks for changing this to a discussion @tiangolo. A much better fit in my opinion. Following up on my experience after I replied to this thread: I do have to say with pythons inner functions you can abstract a lot of things that get at least 80% of what you (or I at least) want to do with classes in this context. By passing the router, my database models and pydantic schemas as parameters to a generalized function that had the inner functions decorated with the router passed to the outer function you can encapsulate a lot of basic functionality such as CRUD operations. The missing feature is the overloading of the defined paths in there but for me that fell in the 20% category (or more like the 5%). With two of these generalized functions I cut down my boilerplate code from 20+ individually defined router files (one per db model) to two generalized router functions. The router files still remained but were reduced to passing the correct parameters to the generalized functions. Adding new features became a breeze by defining them in the generalized functions and they were instantly up and running for all 20+ models as soon as the code got redeployed. |
Beta Was this translation helpful? Give feedback.
-
|
For now, best is to use #8991 (comment). A different solution. from fastapi import FastAPI, APIRouter
class Some:
def __init__(self):
self.name = "some"
class Hello:
def __init__(self, name: str, some):
self.some = some
self.name = name
self.router = APIRouter()
self.router.add_api_route
self.router.add_api_route("/hello", self.hello, methods=["GET"])
def hello(self):
return {
"Hello": self.name,
"self":repr(self),
"some":repr(self.some),
}
# This is essential !!!
def uvicorn_run():
some = Some()
app = FastAPI()
hello = Hello("World", some)
app.include_router(hello.router)
return appBut this is like using starlette. Consider following example. App runs on AWS ECS in docker. App uses postgresql (AWS RDS) as it's storage. Credentials for DB are stored in AWS Parameter Store. What has to be done in order:
With fastapi heavy usage of globals it's impossible:
You'd suggest to use a Use I find FastApi to promote unhealthy habits by promoting heavy use of globals. |
Beta Was this translation helpful? Give feedback.
-
You can use lifespan state now. |
Beta Was this translation helpful? Give feedback.
-
|
But how to pass request object and thus state to function ? |
Beta Was this translation helpful? Give feedback.
-
|
Late, but ... Even you can have class method and bare functions mixed in the same file. And routing as usual:
|
Beta Was this translation helpful? Give feedback.
-
|
My take at this, in case it's useful to anyone (can surely be improved). It allows using real "constructor injected" router classes: import inspect
import logging
from typing import Any, Callable, Optional, Union
from fastapi import APIRouter
logger = logging.getLogger(__name__)
class EndpointDefinition:
def __init__(self, method, *args, **kwargs):
super().__init__()
self.method = method
self.path = args[0] if len(args) > 0 else kwargs.get("path")
self.args = args[1:]
self.kwargs = kwargs
def include_in_router(self, endpoint: Callable[..., Any], router: APIRouter, **defaults_route_args):
kwargs = {**defaults_route_args, **self.kwargs}
router.add_api_route(self.path, endpoint, methods=[self.method], *self.args, **kwargs)
class APIControllerDecorator:
"""
Emulates FastAPI endpoint decorators: Intercepts calls to them and record these in an endpoint method custom
property to replay it later when included in a router.
"""
def __getattribute__(self, __name):
allowed_methods = ["api_route", "delete", "get", "head", "options", "patch", "post", "put", "trace"]
if __name not in allowed_methods:
return NotImplementedError(f"Method {__name} not allowed in APIController")
return APIControllerDecorator._intercept_method(__name)
@staticmethod
def _intercept_method(method: str) -> Callable[..., Any]:
def decorator_factory(*args, **kwargs):
def decorate(endpoint):
if not hasattr(endpoint, "__endpoint_definitions__"):
endpoint.__endpoint_definitions__ = []
endpoint.__endpoint_definitions__.append(EndpointDefinition(method, *args, **kwargs))
return endpoint
return decorate
return decorator_factory
controller: Union[APIRouter, APIControllerDecorator] = APIControllerDecorator()
"""
Endpoint decorator that mimics FastAPI ones.
"""
def router_from_controller(controller: Any, router: Optional[APIRouter] = None, **defaults_route_args) -> APIRouter:
"""
Builds a router from a controller instance (with '@controller.xxx()' annotated endpoints).
"""
router = router or APIRouter()
members = inspect.getmembers(controller, lambda x: hasattr(x, "__endpoint_definitions__"))
if len(members) == 0:
raise ValueError("No endpoints found in provided object")
for k, endpoint in members:
for endpoint_definition in getattr(endpoint, "__endpoint_definitions__"):
logger.debug("Configuring route '%s':", endpoint_definition.path)
logger.debug(" endpoint = %s%s", endpoint.__name__, inspect.signature(endpoint))
endpoint_definition.include_in_router(endpoint, router, **defaults_route_args)
return routerIt mimics and intercepts calls to fastapi decorators (so you have full code completion), and injects what it intercepted later when building the router using You can use it that way: from router import controller
class MyController:
def __init__(self, my_service: MyService):
self._my_service = my_service
@controller.get("/myapi")
async def my_endpoint(self):
await self._my_service.do_some_stuff()
return {"status": "ok"}
...
my_service = MyService()
my_controller = MyController(my_service=my_service)
application = FastAPI()
application.include_router(router_from_controller(my_controller, APIRouter()))Hope it can be useful to anyone. |
Beta Was this translation helpful? Give feedback.
Hey @Kojiro20, thanks for your interest.
I see that your approach is different from
fastapi-utils's class-based views, in that you are actually creating the instances of the classes directly, not expecting FastAPI to do that for you. So I guess it's probably a different use case.About your motivations:
I think you can achieve more or less the same functionality you are showing here with a function, without needing any additional complexity or changes in FastAPI: