tessl/pypi-sanic
A modern Python web server and web framework designed for high performance and speed using async/await syntax.
—
Pending
request-response.mddocs/
Request and Response Handling
Comprehensive request and response objects for handling HTTP communications in Sanic applications. The Request object provides access to all incoming request data, while response functions create properly formatted HTTP responses.
Capabilities
Request Object
The Request object contains all information about the incoming HTTP request, including headers, body data, query parameters, and metadata.
class Request:
"""HTTP request object containing request data and metadata."""
@property
def app(self):
"""Reference to the Sanic application instance."""
@property
def body(self) -> bytes:
"""Raw request body as bytes."""
@property
def content_type(self) -> str:
"""Request content type from headers."""
@property
def json(self):
"""
Request body parsed as JSON.
Returns:
Parsed JSON data or None if not JSON
Raises:
InvalidUsage: If JSON is malformed
"""
@property
def form(self) -> dict:
"""
Form data from request body.
Returns:
Dictionary of form fields
"""
@property
def files(self) -> dict:
"""
Uploaded files from multipart form data.
Returns:
Dictionary of File objects
"""
@property
def args(self) -> dict:
"""
Query string arguments.
Returns:
Dictionary of query parameters
"""
@property
def raw_args(self) -> dict:
"""Raw query string arguments without processing."""
@property
def query_args(self) -> list:
"""
Query arguments as list of tuples.
Returns:
List of (key, value) tuples
"""
@property
def headers(self):
"""
Request headers.
Returns:
Header dictionary
"""
@property
def method(self) -> str:
"""HTTP method (GET, POST, etc.)."""
@property
def ip(self) -> str:
"""Client IP address."""
@property
def forwarded(self) -> dict:
"""Forwarded headers information."""
@property
def remote_addr(self) -> str:
"""Remote client address."""
@property
def scheme(self) -> str:
"""Request scheme (http or https)."""
@property
def host(self) -> str:
"""Request host header."""
@property
def path(self) -> str:
"""Request path without query string."""
@property
def query_string(self) -> str:
"""Raw query string."""
@property
def url(self) -> str:
"""Complete request URL."""
@property
def token(self) -> str:
"""Authorization token from Authorization header."""
@property
def match_info(self) -> dict:
"""Route match information and parameters."""
@property
def stream(self):
"""Request stream for handling large uploads."""
@property
def ctx(self):
"""Request context object for storing custom data."""
def get_args(
self,
key: str,
default=None,
type=str
):
"""
Get query argument with type conversion.
Parameters:
- key: Argument name
- default: Default value if not found
- type: Type conversion function
Returns:
Converted argument value
"""
def get_query_args(self, key: str, default=None):
"""Alias for get_args method."""
def get_form(self, key: str, default=None):
"""
Get form field value.
Parameters:
- key: Form field name
- default: Default value if not found
Returns:
Form field value
"""
def get_files(self, key: str, default=None):
"""
Get uploaded file.
Parameters:
- key: File field name
- default: Default value if not found
Returns:
File object or default
"""
def url_for(self, view_name: str, **kwargs) -> str:
"""
Generate URL for named route.
Parameters:
- view_name: Route name
- **kwargs: URL parameters
Returns:
Generated URL
"""
async def receive_body(self) -> bytes:
"""
Receive request body asynchronously.
Returns:
Request body bytes
"""Response Functions
Factory functions for creating HTTP responses with different content types and behaviors.
def json(
body: Any,
status: int = 200,
headers: Optional[dict[str, str]] = None,
content_type: str = "application/json",
dumps: Optional[Callable[..., AnyStr]] = None,
**kwargs: Any,
):
"""
Create JSON response.
Parameters:
- body: Data to serialize as JSON
- status: HTTP status code
- headers: Additional headers
- content_type: Response content type
- dumps: Custom JSON serializer
Returns:
HTTPResponse with JSON content
"""
def text(
body: str,
status: int = 200,
headers: dict = None,
content_type: str = "text/plain; charset=utf-8",
**kwargs
):
"""
Create plain text response.
Parameters:
- body: Text content
- status: HTTP status code
- headers: Additional headers
- content_type: Response content type
Returns:
HTTPResponse with text content
"""
def html(
body: str,
status: int = 200,
headers: dict = None,
**kwargs
):
"""
Create HTML response.
Parameters:
- body: HTML content
- status: HTTP status code
- headers: Additional headers
Returns:
HTTPResponse with HTML content
"""
def raw(
body: bytes,
status: int = 200,
headers: dict = None,
content_type: str = "application/octet-stream",
**kwargs
):
"""
Create raw bytes response.
Parameters:
- body: Raw bytes content
- status: HTTP status code
- headers: Additional headers
- content_type: Response content type
Returns:
HTTPResponse with raw content
"""
def redirect(
to: str,
status: int = 302,
headers: dict = None,
**kwargs
):
"""
Create redirect response.
Parameters:
- to: Redirect URL
- status: HTTP redirect status code (301, 302, 303, 307, 308)
- headers: Additional headers
Returns:
HTTPResponse with redirect
"""
def file(
location: str,
status: int = 200,
mime_type: str = None,
headers: dict = None,
filename: str = None,
**kwargs
):
"""
Create file response.
Parameters:
- location: File path
- status: HTTP status code
- mime_type: File MIME type
- headers: Additional headers
- filename: Download filename
Returns:
FileResponse for serving files
"""
def empty(
status: int = 204,
headers: Optional[dict[str, str]] = None,
):
"""
Create empty response.
Parameters:
- status: HTTP status code
- headers: Additional headers
Returns:
HTTPResponse with no content
"""HTTPResponse Class
The base response class that all response functions return, providing response manipulation and sending capabilities.
class HTTPResponse:
"""HTTP response object."""
def __init__(
self,
body: bytes = None,
status: int = 200,
headers: dict = None,
content_type: str = None,
**kwargs
):
"""
Initialize HTTP response.
Parameters:
- body: Response body
- status: HTTP status code
- headers: Response headers
- content_type: Content type header
"""
@property
def body(self) -> bytes:
"""Response body as bytes."""
@property
def status(self) -> int:
"""HTTP status code."""
@property
def content_type(self) -> str:
"""Response content type."""
@property
def headers(self):
"""Response headers object."""
@property
def cookies(self):
"""Response cookies."""
async def send(
self,
data: bytes = None,
end_stream: bool = None
):
"""
Send response data.
Parameters:
- data: Data to send
- end_stream: Whether to end the stream
"""
def stream(
self,
request,
write_buffer_size: int = 4096,
**kwargs
):
"""
Stream response data.
Parameters:
- request: Request object
- write_buffer_size: Buffer size for writing
"""Streaming Responses
Handle large responses and real-time data streaming with streaming response capabilities.
class StreamingHTTPResponse(HTTPResponse):
"""Streaming HTTP response for large content."""
def __init__(
self,
streaming_fn,
status: int = 200,
headers: dict = None,
content_type: str = "text/plain",
**kwargs
):
"""
Initialize streaming response.
Parameters:
- streaming_fn: Async function that yields data
- status: HTTP status code
- headers: Response headers
- content_type: Content type
"""
class ResponseStream:
"""Helper for streaming responses."""
async def write(self, data: bytes):
"""
Write data to stream.
Parameters:
- data: Data bytes to write
"""
async def end(self):
"""End the stream."""File Uploads
Handle uploaded files through the request.files interface.
class File:
"""Uploaded file object."""
@property
def name(self) -> str:
"""Original filename."""
@property
def type(self) -> str:
"""File content type."""
@property
def body(self) -> bytes:
"""File content as bytes."""
def save(self, location: str):
"""
Save file to disk.
Parameters:
- location: Save path
"""Usage Examples
Request Data Access
from sanic import Sanic
from sanic.response import json
app = Sanic("MyApp")
@app.route("/api/data", methods=["POST"])
async def handle_data(request):
# Access different types of request data
json_data = request.json
form_data = request.form
files = request.files
query_params = request.args
headers = request.headers
# Process uploaded files
if "upload" in files:
uploaded_file = files["upload"]
uploaded_file.save(f"/uploads/{uploaded_file.name}")
return json({
"received_json": json_data,
"form_fields": list(form_data.keys()),
"query_params": dict(query_params),
"content_type": request.content_type,
"client_ip": request.ip
})Response Generation
from sanic.response import json, text, html, redirect, file
@app.route("/api/users/<user_id:int>")
async def get_user(request, user_id):
user = await fetch_user(user_id)
if not user:
return json({"error": "User not found"}, status=404)
return json({"user": user})
@app.route("/download/<filename>")
async def download_file(request, filename):
file_path = f"/files/{filename}"
return await file(file_path, filename=filename)
@app.route("/redirect")
async def redirect_example(request):
return redirect("/new-location", status=301)
@app.route("/streaming")
async def streaming_example(request):
async def generate_data():
for i in range(1000):
yield f"data chunk {i}\n".encode()
return StreamingHTTPResponse(generate_data)Custom Response Headers and Cookies
from sanic.response import json
@app.route("/api/login", methods=["POST"])
async def login(request):
# Authenticate user
user = await authenticate(request.json)
response = json({"user": user["id"]})
# Set custom headers
response.headers["X-Custom-Header"] = "custom-value"
# Set cookies
response.cookies["session_token"] = user["token"]
response.cookies["session_token"]["httponly"] = True
response.cookies["session_token"]["secure"] = True
response.cookies["session_token"]["max-age"] = 3600
return responseRequest Streaming
@app.route("/upload-large", methods=["POST"], stream=True)
async def upload_large_file(request):
"""Handle large file uploads with streaming."""
async def save_chunks():
with open("/tmp/uploaded_file", "wb") as f:
async for body in request.stream:
f.write(body)
await save_chunks()
return json({"status": "uploaded"})Install with Tessl CLI
npx tessl i tessl/pypi-sanic