Browse Source
- Implement query() method in APIRouter and FastAPI classes - Add comprehensive test suite for QUERY method functionality - Include documentation examples demonstrating QUERY usage - Follow IETF draft-ietf-httpbis-safe-method-w-body-02 specification - Enables complex queries with request bodies for GraphQL-like functionality - Maintains full backward compatibility with existing HTTP methods Resolves: Request for QUERY method support in GitHub issues Addresses: Need for safe HTTP method that allows request bodiespull/13948/head
yogishhg9964
3 days ago
6 changed files with 1620 additions and 0 deletions
@ -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 fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
from typing import Optional |
|||
|
|||
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) |
|||
} |
|||
@ -0,0 +1,256 @@ |
|||
#!/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 fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
from typing import List, Optional, Dict, Any |
|||
|
|||
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 |
|||
} |
|||
} |
|||
@ -0,0 +1,329 @@ |
|||
#!/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 fastapi import FastAPI |
|||
from pydantic import BaseModel |
|||
from typing import List, Optional, Dict, Any |
|||
|
|||
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! |
|||
@ -0,0 +1,222 @@ |
|||
#!/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. |
|||
""" |
|||
|
|||
from fastapi import FastAPI, Depends |
|||
from fastapi.testclient import TestClient |
|||
from pydantic import BaseModel |
|||
from typing import List, Optional |
|||
|
|||
|
|||
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.model_dump()} |
|||
|
|||
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" |
|||
Loading…
Reference in new issue