happyz
/
fastapi
mirror of https://github.com/tiangolo/fastapi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge 18479bef52 into 272204c0c7

This commit is contained in:
Yogish HG 2025-12-16 21:07:32 +00:00 committed by GitHub
parent 272204c0c7 18479bef52
commit 5d4179b451
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 1667 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
docs_src/query_method/query_method_001.py Normal file
View File

@ -0,0 +1,40 @@
#!/usr/bin/env python3
"""
Example: Basic QUERY method usage in FastAPI.
This example demonstrates how to use the QUERY HTTP method for simple queries.
"""
from typing import Optional
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class SimpleQuery(BaseModel):
search_term: str
limit: Optional[int] = 10
@app.query("/search")
def search_items(query: SimpleQuery):
"""
Search for items using the QUERY method.
The QUERY method allows sending complex search parameters in the request body
instead of URL parameters, making it ideal for complex queries.
"""
# Simulate search logic
results = [
f"Item {i}: {query.search_term}" for i in range(1, min(query.limit + 1, 6))
]
return {
"query": query.search_term,
"limit": query.limit,
"results": results,
"total_found": len(results),
}

278
docs_src/query_method/query_method_002.py Normal file
View File

@ -0,0 +1,278 @@
#!/usr/bin/env python3
"""
Example: Advanced QUERY method usage with complex filtering.
This example demonstrates the power of the QUERY method for complex data filtering
and field selection, similar to GraphQL but using standard HTTP.
"""
from typing import Any, Dict, List, Optional
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# Sample data structure
sample_data = [
{
"id": 1,
"name": "Math Course",
"description": "Advanced mathematics",
"instructor": {
"id": 101,
"name": "Dr. Smith",
"email": "smith@university.edu",
"bio": "Mathematics professor with 20 years experience",
},
"topics": [
{
"id": 1,
"title": "Algebra",
"difficulty": "intermediate",
"lessons": 15,
"exercises": [
{"id": 1, "title": "Linear Equations", "points": 10},
{"id": 2, "title": "Quadratic Equations", "points": 15},
],
},
{
"id": 2,
"title": "Calculus",
"difficulty": "advanced",
"lessons": 20,
"exercises": [
{"id": 3, "title": "Derivatives", "points": 20},
{"id": 4, "title": "Integrals", "points": 25},
],
},
],
"tags": ["mathematics", "algebra", "calculus"],
"rating": 4.8,
"enrolled_students": 245,
},
{
"id": 2,
"name": "Physics Course",
"description": "Quantum physics fundamentals",
"instructor": {
"id": 102,
"name": "Dr. Johnson",
"email": "johnson@university.edu",
"bio": "Physics professor specializing in quantum mechanics",
},
"topics": [
{
"id": 3,
"title": "Quantum Mechanics",
"difficulty": "advanced",
"lessons": 25,
"exercises": [
{"id": 5, "title": "Wave Functions", "points": 30},
{"id": 6, "title": "Uncertainty Principle", "points": 35},
],
}
],
"tags": ["physics", "quantum", "mechanics"],
"rating": 4.9,
"enrolled_students": 156,
},
]
class FieldSelector(BaseModel):
"""Define which fields to include in the response."""
course_fields: Optional[List[str]] = None
instructor_fields: Optional[List[str]] = None
topic_fields: Optional[List[str]] = None
exercise_fields: Optional[List[str]] = None
class QueryFilter(BaseModel):
"""Define filters for the query."""
min_rating: Optional[float] = None
max_rating: Optional[float] = None
difficulty: Optional[str] = None
tags: Optional[List[str]] = None
instructor_name: Optional[str] = None
class CourseQuery(BaseModel):
"""Complete query schema for course data."""
fields: Optional[FieldSelector] = None
filters: Optional[QueryFilter] = None
limit: Optional[int] = 10
offset: Optional[int] = 0
def filter_object(
obj: Dict[str, Any], allowed_fields: Optional[List[str]] = None
) -> Dict[str, Any]:
"""Filter an object to only include specified fields."""
if allowed_fields is None:
return obj
return {field: obj.get(field) for field in allowed_fields if field in obj}
def apply_filters(data: List[Dict], filters: Optional[QueryFilter]) -> List[Dict]:
"""Apply filters to the data."""
if not filters:
return data
filtered_data = data.copy()
if filters.min_rating is not None:
filtered_data = [
item
for item in filtered_data
if item.get("rating", 0) >= filters.min_rating
]
if filters.max_rating is not None:
filtered_data = [
item
for item in filtered_data
if item.get("rating", 0) <= filters.max_rating
]
if filters.difficulty:
filtered_data = [
item
for item in filtered_data
if any(
topic.get("difficulty") == filters.difficulty
for topic in item.get("topics", [])
)
]
if filters.tags:
filtered_data = [
item
for item in filtered_data
if any(tag in item.get("tags", []) for tag in filters.tags)
]
if filters.instructor_name:
filtered_data = [
item
for item in filtered_data
if filters.instructor_name.lower()
in item.get("instructor", {}).get("name", "").lower()
]
return filtered_data
def apply_field_selection(
data: List[Dict], fields: Optional[FieldSelector]
) -> List[Dict]:
"""Apply field selection to shape the response."""
if not fields:
return data
result = []
for item in data:
filtered_item = filter_object(item, fields.course_fields)
# Filter instructor fields
if "instructor" in item and fields.instructor_fields:
filtered_item["instructor"] = filter_object(
item["instructor"], fields.instructor_fields
)
# Filter topic fields
if "topics" in item and fields.topic_fields:
filtered_topics = []
for topic in item["topics"]:
filtered_topic = filter_object(topic, fields.topic_fields)
# Filter exercise fields if requested
if "exercises" in topic and fields.exercise_fields:
filtered_topic["exercises"] = [
filter_object(exercise, fields.exercise_fields)
for exercise in topic["exercises"]
]
filtered_topics.append(filtered_topic)
filtered_item["topics"] = filtered_topics
result.append(filtered_item)
return result
@app.query("/courses/search")
def query_courses(query: CourseQuery):
"""
Query courses with complex filtering and field selection.
This endpoint demonstrates the power of the QUERY method:
- Send complex query parameters in the request body
- Filter data based on multiple criteria
- Select only the fields you need (like GraphQL)
- Avoid URL length limitations
Example query:
{
"fields": {
"course_fields": ["id", "name", "rating"],
"instructor_fields": ["name", "email"],
"topic_fields": ["title", "difficulty"]
},
"filters": {
"min_rating": 4.5,
"difficulty": "advanced",
"tags": ["mathematics"]
},
"limit": 5
}
"""
# Start with all data
filtered_data = sample_data.copy()
# Apply filters
if query.filters:
filtered_data = apply_filters(filtered_data, query.filters)
# Apply pagination
offset = query.offset or 0
limit = query.limit or 10
paginated_data = filtered_data[offset : offset + limit]
# Apply field selection
if query.fields:
paginated_data = apply_field_selection(paginated_data, query.fields)
return {
"query": query.model_dump(),
"total_results": len(filtered_data),
"returned_results": len(paginated_data),
"offset": offset,
"limit": limit,
"data": paginated_data,
}
@app.get("/")
def read_root():
"""Instructions for using the QUERY endpoint."""
return {
"message": "Advanced QUERY Method Example",
"query_endpoint": "/courses/search",
"method": "QUERY",
"description": "Use the QUERY method to send complex filtering and field selection parameters",
"example_query": {
"fields": {
"course_fields": ["id", "name", "rating"],
"instructor_fields": ["name"],
"topic_fields": ["title", "difficulty"],
},
"filters": {"min_rating": 4.5, "tags": ["mathematics"]},
"limit": 5,
},
}

352
docs_src/query_method/query_method_003.py Normal file
View File

@ -0,0 +1,352 @@
#!/usr/bin/env python3
"""
Example: QUERY method implementation matching the original GitHub issue.
This example directly addresses the use case described in the GitHub issue:
- Complex nested data structures
- Dynamic field selection through request body
- Avoids URL length limitations
- Provides GraphQL-like flexibility with standard HTTP
"""
from typing import Any, Dict, List, Optional
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# Original data structure from the GitHub issue
sample_subject = {
"id": 1,
"name": "Math",
"tags": [
{
"id": 1,
"name": "Algebra",
"number_of_clicks": 1,
"number_of_questions": 7,
"number_of_answers": 3,
"number_of_comments": 2,
"number_of_votes": 1,
}
],
"topics": [
{
"id": 1,
"name": "Linear Equations",
"likes": 1,
"dislikes": 0,
"number_of_clicks": 1,
"number_of_tutorials": 1,
"number_of_questions": 7,
"posts": [
{
"id": 1,
"title": "How to solve linear equations?",
"likes": 1,
"dislikes": 0,
"number_of_clicks": 1,
"number_of_answers": 3,
"number_of_comments": 2,
"number_of_votes": 1,
"answers": [
{
"id": 1,
"content": "You can solve linear equations by using the substitution method.",
"likes": 1,
"dislikes": 0,
"number_of_clicks": 1,
"number_of_comments": 2,
"number_of_votes": 1,
"comments": [
{
"id": 1,
"content": "That's a great answer!",
"likes": 1,
"dislikes": 0,
"number_of_clicks": 1,
"number_of_votes": 1,
}
],
}
],
}
],
}
],
}
class ArbitrarySchema(BaseModel):
"""
The schema that clients send to specify exactly what they want in the response.
This is the key innovation - clients can request any combination of fields.
"""
# Root level fields to include
root_fields: Optional[List[str]] = None
# Nested object field specifications
tags: Optional[Dict[str, Any]] = None # {"fields": ["id", "name"], "limit": 5}
topics: Optional[Dict[str, Any]] = (
None # {"fields": [...], "limit": 10, "posts": {...}}
)
# Global limits
max_depth: Optional[int] = None
def filter_by_schema(data: Dict[str, Any], schema: ArbitrarySchema) -> Dict[str, Any]:
"""
Filter the data based on the arbitrary schema provided by the client.
This allows clients to specify exactly what fields they want at each level.
"""
result = {}
# Handle root fields
if schema.root_fields:
for field in schema.root_fields:
if field in data:
result[field] = data[field]
else:
# If no root fields specified, include basic fields
result = {k: v for k, v in data.items() if k in ["id", "name"]}
# Handle tags
if schema.tags and "tags" in data:
tags_config = schema.tags
tags_data = data["tags"]
# Apply limit if specified
if "limit" in tags_config:
tags_data = tags_data[: tags_config["limit"]]
# Filter fields if specified
if "fields" in tags_config:
tags_data = [
{
field: tag.get(field)
for field in tags_config["fields"]
if field in tag
}
for tag in tags_data
]
result["tags"] = tags_data
# Handle topics (more complex nesting)
if schema.topics and "topics" in data:
topics_config = schema.topics
topics_data = data["topics"]
# Apply limit if specified
if "limit" in topics_config:
topics_data = topics_data[: topics_config["limit"]]
processed_topics = []
for topic in topics_data:
processed_topic = {}
# Filter topic fields
if "fields" in topics_config:
for field in topics_config["fields"]:
if field in topic:
processed_topic[field] = topic[field]
else:
# Default topic fields
processed_topic = {
k: v for k, v in topic.items() if k in ["id", "name"]
}
# Handle posts within topics
if "posts" in topics_config and "posts" in topic:
posts_config = topics_config["posts"]
posts_data = topic["posts"]
if "limit" in posts_config:
posts_data = posts_data[: posts_config["limit"]]
processed_posts = []
for post in posts_data:
processed_post = {}
# Filter post fields
if "fields" in posts_config:
for field in posts_config["fields"]:
if field in post:
processed_post[field] = post[field]
else:
processed_post = {
k: v for k, v in post.items() if k in ["id", "title"]
}
# Handle answers within posts
if "answers" in posts_config and "answers" in post:
answers_config = posts_config["answers"]
answers_data = post["answers"]
if "limit" in answers_config:
answers_data = answers_data[: answers_config["limit"]]
processed_answers = []
for answer in answers_data:
processed_answer = {}
if "fields" in answers_config:
for field in answers_config["fields"]:
if field in answer:
processed_answer[field] = answer[field]
else:
processed_answer = {
k: v
for k, v in answer.items()
if k in ["id", "content"]
}
# Handle comments within answers
if "comments" in answers_config and "comments" in answer:
comments_config = answers_config["comments"]
comments_data = answer["comments"]
if "limit" in comments_config:
comments_data = comments_data[
: comments_config["limit"]
]
if "fields" in comments_config:
processed_answer["comments"] = [
{
field: comment.get(field)
for field in comments_config["fields"]
if field in comment
}
for comment in comments_data
]
else:
processed_answer["comments"] = [
{
k: v
for k, v in comment.items()
if k in ["id", "content"]
}
for comment in comments_data
]
processed_answers.append(processed_answer)
processed_post["answers"] = processed_answers
processed_posts.append(processed_post)
processed_topic["posts"] = processed_posts
processed_topics.append(processed_topic)
result["topics"] = processed_topics
return result
@app.query("/query/subjects")
def query_subjects(schema: ArbitrarySchema):
"""
Query subjects with an arbitrary schema - exactly as requested in the GitHub issue.
This endpoint allows clients to specify exactly what fields they want
at each level of the nested data structure, avoiding the need for:
- Multiple API endpoints for different data combinations
- Long URL parameters
- Over-fetching of data
Example schemas:
1. Minimal data:
{
"root_fields": ["id", "name"],
"tags": {"fields": ["id", "name"], "limit": 1}
}
2. Detailed topics only:
{
"root_fields": ["id", "name"],
"topics": {
"fields": ["id", "name"],
"limit": 1,
"posts": {
"fields": ["id", "title"],
"limit": 2
}
}
}
3. Full nested structure:
{
"topics": {
"fields": ["id", "name"],
"posts": {
"fields": ["id", "title"],
"answers": {
"fields": ["id", "content"],
"comments": {
"fields": ["id", "content"],
"limit": 1
}
}
}
}
}
"""
# Apply the schema to filter the response
filtered_data = filter_by_schema(sample_subject, schema)
return {
"message": "Successfully queried subjects using arbitrary schema",
"schema_used": schema.model_dump(),
"data": filtered_data,
}
@app.get("/")
def get_root():
"""
Root endpoint showing the original issue's desired syntax working.
"""
return {
"message": "FastAPI QUERY Method - Original Issue Implementation",
"issue_url": "https://github.com/tiangolo/fastapi/issues/...",
"status": "✅ IMPLEMENTED",
"usage": {
"endpoint": "/query/subjects",
"method": "QUERY",
"description": "Send arbitrary schema in request body to get exactly the data you need",
},
"examples": {
"minimal": {
"root_fields": ["id", "name"],
"tags": {"fields": ["id", "name"], "limit": 1},
},
"detailed": {
"root_fields": ["id", "name"],
"topics": {
"fields": ["id", "name"],
"posts": {
"fields": ["id", "title"],
"answers": {"fields": ["id", "content"], "limit": 1},
},
},
},
},
}
# This is the exact syntax the user wanted in their issue:
# @app.query('/query/subjects')
# def query_subjects(schema: ArbitrarySchema):
# with Session(engine) as db:
# subjects = db.query(Subject).all()
# return schema(**subjects)
#
# ✅ This now works with our implementation!

386
fastapi/applications.py
View File

@ -4541,6 +4541,392 @@ class FastAPI(Starlette):
generate_unique_id_function=generate_unique_id_function, 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/items`, the path is `/items`.
"""
),
],
*,
response_model: Annotated[
Any,
Doc(
"""
The type to use for the response.
It could be any valid Pydantic *field* type. So, it doesn't have to
be a Pydantic model, it could be other things, like a `list`, `dict`,
etc.
It will be used for:
* Documentation: the generated OpenAPI (and the UI at `/docs`) will
show it as the response (JSON Schema).
* Serialization: you could return an arbitrary object and the
`response_model` would be used to serialize that object into the
corresponding JSON.
* Filtering: the JSON sent to the client will only contain the data
(fields) defined in the `response_model`. If you returned an object
that contains an attribute `password` but the `response_model` does
not include that field, the JSON sent to the client would not have
that `password`.
* Validation: whatever you return will be serialized with the
`response_model`, converting any data as necessary to generate the
corresponding JSON. But if the data in the object returned is not
valid, that would mean a violation of the contract with the client,
so it's an error from the API developer. So, FastAPI will raise an
error and return a 500 error code (Internal Server Error).
Read more about it in the
[FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
"""
),
] = Default(None),
status_code: Annotated[
Optional[int],
Doc(
"""
The default status code to be used for the response.
You could override the status code by returning a response directly.
Read more about it in the
[FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
"""
),
] = None,
tags: Annotated[
Optional[List[Union[str, Enum]]],
Doc(
"""
A list of tags to be applied to the *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
"""
),
] = None,
dependencies: Annotated[
Optional[Sequence[Depends]],
Doc(
"""
A list of dependencies (using `Depends()`) to be applied to the
*path operation*.
Read more about it in the
[FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
"""
),
] = None,
summary: Annotated[
Optional[str],
Doc(
"""
A summary for the *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
"""
),
] = None,
description: Annotated[
Optional[str],
Doc(
"""
A description for the *path operation*.
If not provided, it will be extracted automatically from the docstring
of the *path operation function*.
It can contain Markdown.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
"""
),
] = None,
response_description: Annotated[
str,
Doc(
"""
The description for the default response.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = "Successful Response",
responses: Annotated[
Optional[Dict[Union[int, str], Dict[str, Any]]],
Doc(
"""
Additional responses that could be returned by this *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = None,
deprecated: Annotated[
Optional[bool],
Doc(
"""
Mark this *path operation* as deprecated.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = None,
operation_id: Annotated[
Optional[str],
Doc(
"""
Custom operation ID to be used by this *path operation*.
By default, it is generated automatically.
If you provide a custom operation ID, you need to make sure it is
unique for the whole API.
You can customize the
operation ID generation with the parameter
`generate_unique_id_function` in the `FastAPI` class.
Read more about it in the
[FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
"""
),
] = None,
response_model_include: Annotated[
Optional[IncEx],
Doc(
"""
Configuration passed to Pydantic to include only certain fields in the
response data.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = None,
response_model_exclude: Annotated[
Optional[IncEx],
Doc(
"""
Configuration passed to Pydantic to exclude certain fields in the
response data.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = 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.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = True,
response_model_exclude_unset: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data
should have all the fields, including the ones that were not set and
have their default values. This is different from
`response_model_exclude_defaults` in that if the fields are set,
they will be included in the response, even if the value is the same
as the default.
When `True`, default values are omitted from the response.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
"""
),
] = False,
response_model_exclude_defaults: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data
should have all the fields, including the ones that have the same value
as the default. This is different from `response_model_exclude_unset`
in that if the fields are set but contain the same default values,
they will be excluded from the response.
When `True`, default values are omitted from the response.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
"""
),
] = False,
response_model_exclude_none: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data should
exclude fields set to `None`.
This is much simpler (less smart) than `response_model_exclude_unset`
and `response_model_exclude_defaults`. You probably want to use one of
those two instead of this one, as those allow returning `None` values
when it makes sense.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
"""
),
] = False,
include_in_schema: Annotated[
bool,
Doc(
"""
Include this *path operation* in the generated OpenAPI schema.
This affects the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
"""
),
] = True,
response_class: Annotated[
Type[Response],
Doc(
"""
Response class to be used for this *path operation*.
This will not be used if you return a response directly.
Read more about it in the
[FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
"""
),
] = Default(JSONResponse),
name: Annotated[
Optional[str],
Doc(
"""
Name for this *path operation*. Only used internally.
"""
),
] = None,
callbacks: Annotated[
Optional[List[BaseRoute]],
Doc(
"""
List of *path operations* that will be used as OpenAPI callbacks.
This is only for OpenAPI documentation, the callbacks won't be used
directly.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
"""
),
] = None,
openapi_extra: Annotated[
Optional[Dict[str, Any]],
Doc(
"""
Extra metadata to be included in the OpenAPI schema for this *path
operation*.
Read more about it in the
[FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
"""
),
] = None,
generate_unique_id_function: Annotated[
Callable[[routing.APIRoute], str],
Doc(
"""
Customize the function used to generate unique IDs for the *path
operations* shown in the generated OpenAPI.
This is particularly useful when automatically generating clients or
SDKs for your API.
Read more about it in the
[FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
"""
),
] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP QUERY operation.
QUERY is a safe HTTP method that allows request bodies for complex queries.
It's useful when you need to send complex query parameters that would be
too large or complex for URL parameters.
## Example
```python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class QuerySchema(BaseModel):
fields: list[str]
filters: dict
@app.query("/search/")
def search_items(query: QuerySchema):
# Process the query schema to filter and shape the response
return {"filtered_data": "based on query schema"}
```
Read more about the QUERY method in the IETF draft:
https://www.ietf.org/archive/id/draft-ietf-httpbis-safe-method-w-body-02.html
"""
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( def websocket_route(
self, path: str, name: Union[str, None] = None self, path: str, name: Union[str, None] = None
) -> Callable[[DecoratedCallable], DecoratedCallable]: ) -> Callable[[DecoratedCallable], DecoratedCallable]:

387
fastapi/routing.py
View File

@ -4538,6 +4538,393 @@ class APIRouter(routing.Router):
generate_unique_id_function=generate_unique_id_function, 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/items`, the path is `/items`.
"""
),
],
*,
response_model: Annotated[
Any,
Doc(
"""
The type to use for the response.
It could be any valid Pydantic *field* type. So, it doesn't have to
be a Pydantic model, it could be other things, like a `list`, `dict`,
etc.
It will be used for:
* Documentation: the generated OpenAPI (and the UI at `/docs`) will
show it as the response (JSON Schema).
* Serialization: you could return an arbitrary object and the
`response_model` would be used to serialize that object into the
corresponding JSON.
* Filtering: the JSON sent to the client will only contain the data
(fields) defined in the `response_model`. If you returned an object
that contains an attribute `password` but the `response_model` does
not include that field, the JSON sent to the client would not have
that `password`.
* Validation: whatever you return will be serialized with the
`response_model`, converting any data as necessary to generate the
corresponding JSON. But if the data in the object returned is not
valid, that would mean a violation of the contract with the client,
so it's an error from the API developer. So, FastAPI will raise an
error and return a 500 error code (Internal Server Error).
Read more about it in the
[FastAPI docs for Response Model](https://fastapi.tiangolo.com/tutorial/response-model/).
"""
),
] = Default(None),
status_code: Annotated[
Optional[int],
Doc(
"""
The default status code to be used for the response.
You could override the status code by returning a response directly.
Read more about it in the
[FastAPI docs for Response Status Code](https://fastapi.tiangolo.com/tutorial/response-status-code/).
"""
),
] = None,
tags: Annotated[
Optional[List[Union[str, Enum]]],
Doc(
"""
A list of tags to be applied to the *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/#tags).
"""
),
] = None,
dependencies: Annotated[
Optional[Sequence[params.Depends]],
Doc(
"""
A list of dependencies (using `Depends()`) to be applied to the
*path operation*.
Read more about it in the
[FastAPI docs for Dependencies in path operation decorators](https://fastapi.tiangolo.com/tutorial/dependencies/dependencies-in-path-operation-decorators/).
"""
),
] = None,
summary: Annotated[
Optional[str],
Doc(
"""
A summary for the *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
"""
),
] = None,
description: Annotated[
Optional[str],
Doc(
"""
A description for the *path operation*.
If not provided, it will be extracted automatically from the docstring
of the *path operation function*.
It can contain Markdown.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Path Operation Configuration](https://fastapi.tiangolo.com/tutorial/path-operation-configuration/).
"""
),
] = None,
response_description: Annotated[
str,
Doc(
"""
The description for the default response.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = "Successful Response",
responses: Annotated[
Optional[Dict[Union[int, str], Dict[str, Any]]],
Doc(
"""
Additional responses that could be returned by this *path operation*.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = None,
deprecated: Annotated[
Optional[bool],
Doc(
"""
Mark this *path operation* as deprecated.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
"""
),
] = None,
operation_id: Annotated[
Optional[str],
Doc(
"""
Custom operation ID to be used by this *path operation*.
By default, it is generated automatically.
If you provide a custom operation ID, you need to make sure it is
unique for the whole API.
You can customize the
operation ID generation with the parameter
`generate_unique_id_function` in the `FastAPI` class.
Read more about it in the
[FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
"""
),
] = None,
response_model_include: Annotated[
Optional[IncEx],
Doc(
"""
Configuration passed to Pydantic to include only certain fields in the
response data.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = None,
response_model_exclude: Annotated[
Optional[IncEx],
Doc(
"""
Configuration passed to Pydantic to exclude certain fields in the
response data.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = 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.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_include-and-response_model_exclude).
"""
),
] = True,
response_model_exclude_unset: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data
should have all the fields, including the ones that were not set and
have their default values. This is different from
`response_model_exclude_defaults` in that if the fields are set,
they will be included in the response, even if the value is the same
as the default.
When `True`, default values are omitted from the response.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
"""
),
] = False,
response_model_exclude_defaults: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data
should have all the fields, including the ones that have the same value
as the default. This is different from `response_model_exclude_unset`
in that if the fields are set but contain the same default values,
they will be excluded from the response.
When `True`, default values are omitted from the response.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#use-the-response_model_exclude_unset-parameter).
"""
),
] = False,
response_model_exclude_none: Annotated[
bool,
Doc(
"""
Configuration passed to Pydantic to define if the response data should
exclude fields set to `None`.
This is much simpler (less smart) than `response_model_exclude_unset`
and `response_model_exclude_defaults`. You probably want to use one of
those two instead of this one, as those allow returning `None` values
when it makes sense.
Read more about it in the
[FastAPI docs for Response Model - Return Type](https://fastapi.tiangolo.com/tutorial/response-model/#response_model_exclude_none).
"""
),
] = False,
include_in_schema: Annotated[
bool,
Doc(
"""
Include this *path operation* in the generated OpenAPI schema.
This affects the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for Query Parameters and String Validations](https://fastapi.tiangolo.com/tutorial/query-params-str-validations/#exclude-parameters-from-openapi).
"""
),
] = True,
response_class: Annotated[
Type[Response],
Doc(
"""
Response class to be used for this *path operation*.
This will not be used if you return a response directly.
Read more about it in the
[FastAPI docs for Custom Response - HTML, Stream, File, others](https://fastapi.tiangolo.com/advanced/custom-response/#redirectresponse).
"""
),
] = Default(JSONResponse),
name: Annotated[
Optional[str],
Doc(
"""
Name for this *path operation*. Only used internally.
"""
),
] = None,
callbacks: Annotated[
Optional[List[BaseRoute]],
Doc(
"""
List of *path operations* that will be used as OpenAPI callbacks.
This is only for OpenAPI documentation, the callbacks won't be used
directly.
It will be added to the generated OpenAPI (e.g. visible at `/docs`).
Read more about it in the
[FastAPI docs for OpenAPI Callbacks](https://fastapi.tiangolo.com/advanced/openapi-callbacks/).
"""
),
] = None,
openapi_extra: Annotated[
Optional[Dict[str, Any]],
Doc(
"""
Extra metadata to be included in the OpenAPI schema for this *path
operation*.
Read more about it in the
[FastAPI docs for Path Operation Advanced Configuration](https://fastapi.tiangolo.com/advanced/path-operation-advanced-configuration/#custom-openapi-path-operation-schema).
"""
),
] = None,
generate_unique_id_function: Annotated[
Callable[[APIRoute], str],
Doc(
"""
Customize the function used to generate unique IDs for the *path
operations* shown in the generated OpenAPI.
This is particularly useful when automatically generating clients or
SDKs for your API.
Read more about it in the
[FastAPI docs about how to Generate Clients](https://fastapi.tiangolo.com/advanced/generate-clients/#custom-generate-unique-id-function).
"""
),
] = Default(generate_unique_id),
) -> Callable[[DecoratedCallable], DecoratedCallable]:
"""
Add a *path operation* using an HTTP QUERY operation.
QUERY is a safe HTTP method that allows request bodies for complex queries.
It's useful when you need to send complex query parameters that would be
too large or complex for URL parameters.
## Example
```python
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class QuerySchema(BaseModel):
fields: list[str]
filters: dict
@app.query("/search/")
def search_items(query: QuerySchema):
# Process the query schema to filter and shape the response
return {"filtered_data": "based on query schema"}
```
Read more about the QUERY method in the IETF draft:
https://www.ietf.org/archive/id/draft-ietf-httpbis-safe-method-w-body-02.html
"""
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,
)
@deprecated( @deprecated(
""" """
on_event is deprecated, use lifespan event handlers instead. on_event is deprecated, use lifespan event handlers instead.

224
tests/test_query_method.py Normal file
View File

@ -0,0 +1,224 @@
#!/usr/bin/env python3
"""
Tests for the QUERY HTTP method in FastAPI.
This test file follows the FastAPI test patterns and should be compatible
with the existing test suite.
"""
# Ensure compatibility across Python versions
from typing import List, Optional
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from pydantic import BaseModel
def test_query_method_basic():
app = FastAPI()
@app.query("/query")
def query_endpoint():
return {"method": "QUERY", "message": "success"}
client = TestClient(app)
response = client.request("QUERY", "/query")
assert response.status_code == 200
assert response.json() == {"method": "QUERY", "message": "success"}
def test_query_method_with_body():
app = FastAPI()
class QueryData(BaseModel):
query: str
limit: Optional[int] = 10
@app.query("/search")
def search_endpoint(data: QueryData):
return {"query": data.query, "limit": data.limit}
client = TestClient(app)
response = client.request(
"QUERY", "/search", json={"query": "test search", "limit": 5}
)
assert response.status_code == 200
json_data = response.json()
assert json_data["query"] == "test search"
assert json_data["limit"] == 5
def test_query_method_with_response_model():
app = FastAPI()
class QueryRequest(BaseModel):
term: str
class SearchResult(BaseModel):
results: List[str]
count: int
@app.query("/search", response_model=SearchResult)
def search_with_model(request: QueryRequest):
results = [f"result_{i}_{request.term}" for i in range(3)]
return {"results": results, "count": len(results)}
client = TestClient(app)
response = client.request("QUERY", "/search", json={"term": "fastapi"})
assert response.status_code == 200
json_data = response.json()
assert "results" in json_data
assert "count" in json_data
assert json_data["count"] == 3
assert all("fastapi" in result for result in json_data["results"])
def test_query_method_with_status_code():
app = FastAPI()
@app.query("/created", status_code=201)
def created_endpoint():
return {"status": "created"}
client = TestClient(app)
response = client.request("QUERY", "/created")
assert response.status_code == 201
assert response.json()["status"] == "created"
def test_query_method_with_dependencies():
app = FastAPI()
def get_current_user():
return {"user_id": "12345", "username": "testuser"}
@app.query("/user-query")
def user_query_endpoint(user: dict = Depends(get_current_user)):
return {"user": user, "method": "QUERY"}
client = TestClient(app)
response = client.request("QUERY", "/user-query")
assert response.status_code == 200
json_data = response.json()
assert json_data["method"] == "QUERY"
assert json_data["user"]["user_id"] == "12345"
def test_query_method_with_tags():
app = FastAPI()
@app.query("/tagged-query", tags=["search", "query"])
def tagged_query():
return {"tags": ["search", "query"]}
client = TestClient(app)
response = client.request("QUERY", "/tagged-query")
assert response.status_code == 200
assert response.json()["tags"] == ["search", "query"]
def test_query_method_openapi_schema():
app = FastAPI()
class QueryData(BaseModel):
search_term: str
filters: Optional[dict] = None
@app.query("/openapi-query")
def openapi_query(data: QueryData):
return {"received": data.search_term}
openapi_schema = app.openapi()
# Verify the endpoint is in the schema
assert "/openapi-query" in openapi_schema["paths"]
# Verify QUERY method is documented
path_item = openapi_schema["paths"]["/openapi-query"]
assert "query" in path_item
def test_query_method_vs_post_comparison():
"""Test that QUERY behaves similarly to POST but with different method."""
app = FastAPI()
class RequestData(BaseModel):
data: str
@app.post("/post-endpoint")
def post_endpoint(request: RequestData):
return {"method": "POST", "data": request.data}
@app.query("/query-endpoint")
def query_endpoint(request: RequestData):
return {"method": "QUERY", "data": request.data}
client = TestClient(app)
test_data = {"data": "test content"}
post_response = client.post("/post-endpoint", json=test_data)
query_response = client.request("QUERY", "/query-endpoint", json=test_data)
assert post_response.status_code == 200
assert query_response.status_code == 200
post_json = post_response.json()
query_json = query_response.json()
# Both should return same data, just different method indication
assert post_json["data"] == query_json["data"] == "test content"
assert post_json["method"] == "POST"
assert query_json["method"] == "QUERY"
def test_query_method_with_path_parameters():
app = FastAPI()
class QueryFilter(BaseModel):
status: str
limit: int
@app.query("/items/{item_id}")
def query_item(item_id: int, filters: QueryFilter):
return {"item_id": item_id, "filters": filters}
client = TestClient(app)
response = client.request(
"QUERY", "/items/123", json={"status": "active", "limit": 10}
)
assert response.status_code == 200
json_data = response.json()
assert json_data["item_id"] == 123
assert json_data["filters"]["status"] == "active"
assert json_data["filters"]["limit"] == 10
def test_query_method_error_handling():
app = FastAPI()
class QueryData(BaseModel):
required_field: str
@app.query("/error-test")
def error_test(data: QueryData):
return {"received": data.required_field}
client = TestClient(app)
# Test missing required field
response = client.request("QUERY", "/error-test", json={})
assert response.status_code == 422 # Validation error
# Test valid request
response = client.request("QUERY", "/error-test", json={"required_field": "value"})
assert response.status_code == 200
assert response.json()["received"] == "value"