Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

♻️ Allow callabe in dependant for get_request_handler #11508

Open
wants to merge 9 commits into
base: master
Choose a base branch
from
6 changes: 5 additions & 1 deletion fastapi/routing.py
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,11 @@ def get_request_handler(
dependency_overrides_provider: Optional[Any] = None,
) -> Callable[[Request], Coroutine[Any, Any, Response]]:
assert dependant.call is not None, "dependant.call must be a function"
is_coroutine = asyncio.iscoroutinefunction(dependant.call)
is_coroutine = (
asyncio.iscoroutinefunction(dependant.call)
or callable(dependant.call)
and inspect.iscoroutinefunction(dependant.call.__call__) # type: ignore[operator]
)
is_body_form = body_field and isinstance(body_field.field_info, params.Form)
if isinstance(response_class, DefaultPlaceholder):
actual_response_class: Type[Response] = response_class.value
Expand Down
65 changes: 65 additions & 0 deletions tests/test_endpoint_decorator.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
from functools import update_wrapper
from typing import Any, Callable

from fastapi import Depends, FastAPI
from fastapi.routing import APIRoute
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from fastapi.testclient import TestClient
from starlette.exceptions import HTTPException


class EndpointWrapper(Callable[..., Any]):
def __init__(self, endpoint: Callable[..., Any]):
self.endpoint = endpoint
self.protected = False
update_wrapper(self, endpoint)

async def __call__(self, *args, **kwargs):
return await self.endpoint(*args, **kwargs)


def dummy_secruity_check(token: HTTPAuthorizationCredentials = Depends(HTTPBearer())):
if token.credentials != "fake-token":
raise HTTPException(status_code=401, detail="Unauthorized")


def protect(endpoint: Callable[..., Any]):
if not isinstance(endpoint, EndpointWrapper):
endpoint = EndpointWrapper(endpoint)
endpoint.protected = True
return endpoint


class CustomAPIRoute(APIRoute):
def __init__(
self, path: str, endpoint: Callable[..., Any], dependencies, **kwargs
) -> None:
if isinstance(endpoint, EndpointWrapper) and endpoint.protected:
dependencies.append(Depends(dummy_secruity_check))
super().__init__(path, endpoint, dependencies=dependencies, **kwargs)


app = FastAPI()

app.router.route_class = CustomAPIRoute


@app.get("/protected")
@protect
async def protected_route():
return {"message": "This is a protected route"}


client = TestClient(app)


def test_protected_route():
response = client.get("/protected")
assert response.status_code == 403

response = client.get("/protected", headers={"Authorization": "Bearer some-token"})
assert response.status_code == 401

response = client.get("/protected", headers={"Authorization": "Bearer fake-token"})
assert response.status_code == 200
assert response.json() == {"message": "This is a protected route"}