From 94bde339dd90d7ded9bc1232abb3de12f0c7745c Mon Sep 17 00:00:00 2001 From: WarisAmir Date: Wed, 8 Jul 2026 12:10:06 +0530 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20QUERY=20HTTP=20method=20suppo?= =?UTF-8?q?rt=20to=20FastAPI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Introduce a new custom HTTP method called QUERY for structured query operations. Features: - Full integration with FastAPI routing system - Available on both FastAPI and APIRouter classes - Complete parity with existing HTTP methods - OpenAPI documentation support - Response model validation - Parameter dependency injection - 9 passing tests Examples: @app.query("/search") def search(q: str, limit: int = 10): return {"results": [], "query": q} # Call with: client.request("QUERY", "/search?q=python") --- fastapi/applications.py | 240 +++++++++++++++++++++++++++++++ fastapi/routing.py | 246 ++++++++++++++++++++++++++++++++ tests/test_query_http_method.py | 185 ++++++++++++++++++++++++ 3 files changed, 671 insertions(+) create mode 100644 tests/test_query_http_method.py diff --git a/fastapi/applications.py b/fastapi/applications.py index 56e1a3e60..104c752ec 100644 --- a/fastapi/applications.py +++ b/fastapi/applications.py @@ -4636,6 +4636,246 @@ class FastAPI(Starlette): generate_unique_id_function=generate_unique_id_function, ) + def query( + self, + path: Annotated[ + str, + Doc( + """ + The URL path to be used for this *path operation*. + """ + ), + ], + *, + response_model: Annotated[ + Any, + Doc( + """ + The type to use for the response. + """ + ), + ] = Default(None), + status_code: Annotated[ + int | None, + Doc( + """ + The default status code to be used for the response. + """ + ), + ] = None, + tags: Annotated[ + list[str | Enum] | None, + Doc( + """ + A list of tags to be applied to the *path operation*. + """ + ), + ] = None, + dependencies: Annotated[ + Sequence[params.Depends] | None, + Doc( + """ + A list of dependencies (using `Depends()`) to be applied to the + *path operation*. + """ + ), + ] = None, + summary: Annotated[ + str | None, + Doc( + """ + A summary for the *path operation*. + """ + ), + ] = None, + description: Annotated[ + str | None, + Doc( + """ + A description for the *path operation*. + """ + ), + ] = None, + response_description: Annotated[ + str, + Doc( + """ + The description for the default response. + """ + ), + ] = "Successful Response", + responses: Annotated[ + dict[int | str, dict[str, Any]] | None, + Doc( + """ + Additional responses that could be returned by this *path operation*. + """ + ), + ] = None, + deprecated: Annotated[ + bool | None, + Doc( + """ + Mark this *path operation* as deprecated. + """ + ), + ] = None, + operation_id: Annotated[ + str | None, + Doc( + """ + Custom operation ID to be used by this *path operation*. + """ + ), + ] = None, + response_model_include: Annotated[ + IncEx | None, + Doc( + """ + Configuration passed to Pydantic to include only certain fields in the + response data. + """ + ), + ] = None, + response_model_exclude: Annotated[ + IncEx | None, + Doc( + """ + Configuration passed to Pydantic to exclude certain fields in the + response data. + """ + ), + ] = None, + response_model_by_alias: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should be serialized by alias when an alias is used. + """ + ), + ] = True, + response_model_exclude_unset: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields that were not explicitly set. + """ + ), + ] = False, + response_model_exclude_defaults: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields that are set to their default value. + """ + ), + ] = False, + response_model_exclude_none: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields set to `None`. + """ + ), + ] = False, + include_in_schema: Annotated[ + bool, + Doc( + """ + By default, this *path operation* is included in the generated OpenAPI + schema. + """ + ), + ] = True, + response_class: Annotated[ + type[Response], + Doc( + """ + Response class to use for this *path operation*. + """ + ), + ] = Default(JSONResponse), + name: Annotated[ + str | None, + Doc( + """ + The name for this *path operation*. Only used internally, by FastAPI. + """ + ), + ] = None, + callbacks: Annotated[ + list[BaseRoute] | None, + Doc( + """ + OpenAPI callbacks that should be triggered by this *path operation*. + """ + ), + ] = None, + openapi_extra: Annotated[ + dict[str, Any] | None, + Doc( + """ + Extra metadata to be added to the OpenAPI schema for this *path + operation*. + """ + ), + ] = None, + generate_unique_id_function: Annotated[ + Callable[[APIRoute], str], + Doc( + """ + A callable to generate a unique ID for this *path operation*. + """ + ), + ] = Default(generate_unique_id), + ) -> Callable[[DecoratedCallable], DecoratedCallable]: + """ + Add a *path operation* using an HTTP QUERY operation. + + QUERY is a custom HTTP method for structured query operations. + + ## Example + + ```python + from fastapi import FastAPI + + app = FastAPI() + + @app.query("/search") + def search_items(q: str): + return [{"name": "Item " + q}] + ``` + """ + return self.router.query( + path, + response_model=response_model, + status_code=status_code, + tags=tags, + dependencies=dependencies, + summary=summary, + description=description, + response_description=response_description, + responses=responses, + deprecated=deprecated, + operation_id=operation_id, + response_model_include=response_model_include, + response_model_exclude=response_model_exclude, + response_model_by_alias=response_model_by_alias, + response_model_exclude_unset=response_model_exclude_unset, + response_model_exclude_defaults=response_model_exclude_defaults, + response_model_exclude_none=response_model_exclude_none, + include_in_schema=include_in_schema, + response_class=response_class, + name=name, + callbacks=callbacks, + openapi_extra=openapi_extra, + generate_unique_id_function=generate_unique_id_function, + ) + def websocket_route( self, path: str, name: str | None = None ) -> Callable[[DecoratedCallable], DecoratedCallable]: diff --git a/fastapi/routing.py b/fastapi/routing.py index c442b122b..09db98c76 100644 --- a/fastapi/routing.py +++ b/fastapi/routing.py @@ -6309,6 +6309,252 @@ class APIRouter(routing.Router): generate_unique_id_function=generate_unique_id_function, ) + def query( + self, + path: Annotated[ + str, + Doc( + """ + The URL path to be used for this *path operation*. + + For example, in `http://example.com/search`, the path is `/search`. + """ + ), + ], + *, + response_model: Annotated[ + Any, + Doc( + """ + The type to use for the response. + """ + ), + ] = Default(None), + status_code: Annotated[ + int | None, + Doc( + """ + The default status code to be used for the response. + """ + ), + ] = None, + tags: Annotated[ + list[str | Enum] | None, + Doc( + """ + A list of tags to be applied to the *path operation*. + """ + ), + ] = None, + dependencies: Annotated[ + Sequence[params.Depends] | None, + Doc( + """ + A list of dependencies (using `Depends()`) to be applied to the + *path operation*. + """ + ), + ] = None, + summary: Annotated[ + str | None, + Doc( + """ + A summary for the *path operation*. + """ + ), + ] = None, + description: Annotated[ + str | None, + Doc( + """ + A description for the *path operation*. + """ + ), + ] = None, + response_description: Annotated[ + str, + Doc( + """ + The description for the default response. + """ + ), + ] = "Successful Response", + responses: Annotated[ + dict[int | str, dict[str, Any]] | None, + Doc( + """ + Additional responses that could be returned by this *path operation*. + """ + ), + ] = None, + deprecated: Annotated[ + bool | None, + Doc( + """ + Mark this *path operation* as deprecated. + """ + ), + ] = None, + operation_id: Annotated[ + str | None, + Doc( + """ + Custom operation ID to be used by this *path operation*. + """ + ), + ] = None, + response_model_include: Annotated[ + IncEx | None, + Doc( + """ + Configuration passed to Pydantic to include only certain fields in the + response data. + """ + ), + ] = None, + response_model_exclude: Annotated[ + IncEx | None, + Doc( + """ + Configuration passed to Pydantic to exclude certain fields in the + response data. + """ + ), + ] = None, + response_model_by_alias: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should be serialized by alias when an alias is used. + """ + ), + ] = True, + response_model_exclude_unset: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields that were not explicitly set. + """ + ), + ] = False, + response_model_exclude_defaults: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields that are set to their default value. + """ + ), + ] = False, + response_model_exclude_none: Annotated[ + bool, + Doc( + """ + Configuration passed to Pydantic to define if the response model + should exclude fields set to `None`. + """ + ), + ] = False, + include_in_schema: Annotated[ + bool, + Doc( + """ + By default, this *path operation* is included in the generated OpenAPI + schema. + """ + ), + ] = True, + response_class: Annotated[ + type[Response], + Doc( + """ + Response class to use for this *path operation*. + """ + ), + ] = Default(JSONResponse), + name: Annotated[ + str | None, + Doc( + """ + The name for this *path operation*. Only used internally, by FastAPI. + """ + ), + ] = None, + callbacks: Annotated[ + list[BaseRoute] | None, + Doc( + """ + OpenAPI callbacks that should be triggered by this *path operation*. + """ + ), + ] = None, + openapi_extra: Annotated[ + dict[str, Any] | None, + Doc( + """ + Extra metadata to be added to the OpenAPI schema for this *path + operation*. + """ + ), + ] = None, + generate_unique_id_function: Annotated[ + Callable[[APIRoute], str], + Doc( + """ + A callable to generate a unique ID for this *path operation*. + """ + ), + ] = Default(generate_unique_id), + ) -> Callable[[DecoratedCallable], DecoratedCallable]: + """ + Add a *path operation* using an HTTP QUERY operation. + + QUERY is a custom HTTP method for structured query operations. + + ## Example + + ```python + from fastapi import APIRouter, FastAPI + + app = FastAPI() + router = APIRouter() + + @router.query("/search") + def search_items(q: str): + return [{"name": "Item " + q}] + + app.include_router(router) + ``` + """ + return self.api_route( + path=path, + response_model=response_model, + status_code=status_code, + tags=tags, + dependencies=dependencies, + summary=summary, + description=description, + response_description=response_description, + responses=responses, + deprecated=deprecated, + methods=["QUERY"], + operation_id=operation_id, + response_model_include=response_model_include, + response_model_exclude=response_model_exclude, + response_model_by_alias=response_model_by_alias, + response_model_exclude_unset=response_model_exclude_unset, + response_model_exclude_defaults=response_model_exclude_defaults, + response_model_exclude_none=response_model_exclude_none, + include_in_schema=include_in_schema, + response_class=response_class, + name=name, + callbacks=callbacks, + openapi_extra=openapi_extra, + generate_unique_id_function=generate_unique_id_function, + ) + # TODO: remove this once the lifespan (or alternative) interface is improved async def _startup(self) -> None: """ diff --git a/tests/test_query_http_method.py b/tests/test_query_http_method.py new file mode 100644 index 000000000..503cde9ea --- /dev/null +++ b/tests/test_query_http_method.py @@ -0,0 +1,185 @@ +""" +Tests for the QUERY HTTP method in FastAPI. + +The QUERY method is a custom HTTP method designed for structured query operations. +""" + +import pytest +from fastapi import APIRouter, FastAPI +from starlette.testclient import TestClient + + +class TestQueryHttpMethod: + """Tests for the QUERY HTTP method.""" + + @pytest.fixture + def app(self): + """Create a test FastAPI application.""" + app = FastAPI() + + @app.query("/search") + def search(q: str, limit: int = 10): + """Simple search endpoint using QUERY method.""" + return { + "query": q, + "limit": limit, + "items": [{"id": 1, "name": f"Result for {q}"}], + } + + return app + + def test_query_method_basic(self, app): + """Test basic QUERY method request.""" + client = TestClient(app) + response = client.request("QUERY", "/search?q=python") + assert response.status_code == 200 + data = response.json() + assert data["query"] == "python" + assert data["limit"] == 10 + + def test_query_method_with_params(self, app): + """Test QUERY method with multiple parameters.""" + client = TestClient(app) + response = client.request("QUERY", "/search?q=fastapi&limit=20") + assert response.status_code == 200 + data = response.json() + assert data["query"] == "fastapi" + assert data["limit"] == 20 + + def test_query_method_default_params(self, app): + """Test QUERY method uses default parameter values.""" + client = TestClient(app) + response = client.request("QUERY", "/search?q=default") + assert response.status_code == 200 + data = response.json() + assert data["limit"] == 10 + + @pytest.fixture + def router_app(self): + """Test QUERY method with APIRouter.""" + app = FastAPI() + router = APIRouter() + + @router.query("/items/search") + def search_items(name: str, skip: int = 0): + """Search for items.""" + return { + "search_term": name, + "skip": skip, + "found_items": [{"id": 1, "name": name}], + } + + app.include_router(router) + return app + + def test_query_method_with_router(self, router_app): + """Test QUERY method with APIRouter.""" + client = TestClient(router_app) + response = client.request("QUERY", "/items/search?name=widget&skip=5") + assert response.status_code == 200 + data = response.json() + assert data["search_term"] == "widget" + assert data["skip"] == 5 + + @pytest.fixture + def tagged_app(self): + """Test QUERY method with tags and metadata.""" + app = FastAPI() + + @app.query( + "/search", + tags=["search"], + summary="Search endpoint", + description="Search for items using QUERY method", + ) + def search(q: str): + return {"query": q, "results": []} + + return app + + def test_query_method_with_tags(self, tagged_app): + """Test QUERY method request succeeds.""" + client = TestClient(tagged_app) + response = client.request("QUERY", "/search?q=tagged") + assert response.status_code == 200 + assert response.json()["query"] == "tagged" + + def test_query_method_openapi_documentation(self, tagged_app): + """Test that QUERY method appears in OpenAPI documentation.""" + client = TestClient(tagged_app) + response = client.get("/openapi.json") + assert response.status_code == 200 + openapi = response.json() + assert "/search" in openapi["paths"] + + @pytest.fixture + def multiple_query_methods(self): + """Test multiple QUERY endpoints.""" + app = FastAPI() + + @app.query("/users/search") + def search_users(name: str): + return {"type": "users", "query": name} + + @app.query("/posts/search") + def search_posts(title: str): + return {"type": "posts", "query": title} + + return app + + def test_multiple_query_endpoints(self, multiple_query_methods): + """Test multiple QUERY method endpoints.""" + client = TestClient(multiple_query_methods) + + # Test first endpoint + response1 = client.request("QUERY", "/users/search?name=alice") + assert response1.status_code == 200 + assert response1.json()["type"] == "users" + + # Test second endpoint + response2 = client.request("QUERY", "/posts/search?title=hello") + assert response2.status_code == 200 + assert response2.json()["type"] == "posts" + + +class TestQueryMethodIntegration: + """Test integration with FastAPI features.""" + + @pytest.fixture + def featured_app(self): + """App with various FastAPI features.""" + app = FastAPI() + + @app.query("/search", status_code=200, deprecated=False) + def search(q: str, limit: int = 10): + """Search endpoint with custom config.""" + return {"q": q, "limit": limit} + + @app.get("/items") + def list_items(): + """GET endpoint for comparison.""" + return {"items": []} + + return app + + def test_query_and_get_coexist(self, featured_app): + """Test that QUERY and GET methods can coexist.""" + client = TestClient(featured_app) + + # GET should work + get_response = client.get("/items") + assert get_response.status_code == 200 + + # QUERY should also work + query_response = client.request("QUERY", "/search?q=test") + assert query_response.status_code == 200 + + def test_query_method_with_status_code(self, featured_app): + """Test QUERY method respects status_code parameter.""" + client = TestClient(featured_app) + response = client.request("QUERY", "/search?q=test") + assert response.status_code == 200 + + +if __name__ == "__main__": + pytest.main([__file__, "-v"])