tessl/pypi-fastapi-slim

FastAPI framework, high performance, easy to learn, fast to code, ready for production - slim version without standard dependencies

—

Pending

Overview

Eval results

Files

Request and Response Handling

Name: tessl/pypi-fastapi-slim
Author: tessl

FastAPI provides comprehensive request and response objects for accessing HTTP request data and creating custom responses. The Request object gives access to all request information, while Response classes allow fine-grained control over HTTP responses.

Capabilities

Request Object

Object for accessing all HTTP request information including headers, body, query parameters, and connection details.

class Request:
    """
    HTTP request object providing access to all request data.
    
    Attributes (read-only):
    - method: HTTP method (GET, POST, etc.)
    - url: Complete URL object with components
    - headers: HTTP headers as case-insensitive mapping
    - query_params: Query parameters as mapping
    - path_params: Path parameters extracted from URL pattern
    - cookies: HTTP cookies as mapping
    - client: Client connection information (host, port)
    - session: Session data (if session middleware is enabled)
    - state: Application state for storing data across middleware
    """
    
    method: str
    url: URL
    headers: Headers
    query_params: QueryParams
    path_params: Dict[str, Any]
    cookies: Dict[str, str]
    client: Optional[Address]
    session: Dict[str, Any]
    state: State
    
    async def body(self) -> bytes:
        """
        Get raw request body as bytes.
        
        Returns:
        Complete request body as bytes
        
        Note: Can only be called once per request
        """
    
    async def json(self) -> Any:
        """
        Parse request body as JSON.
        
        Returns:
        Parsed JSON data (dict, list, or primitive)
        
        Raises:
        JSONDecodeError: If body is not valid JSON
        """
    
    async def form(self) -> FormData:
        """
        Parse request body as form data.
        
        Returns:
        FormData object with form fields and files
        
        Note: Handles both URL-encoded and multipart form data
        """
    
    def stream(self) -> AsyncGenerator[bytes, None]:
        """
        Get request body as async stream.
        
        Returns:
        Async generator yielding body chunks
        
        Note: Useful for large request bodies
        """
    
    def scope(self) -> Dict[str, Any]:
        """
        Get ASGI scope dictionary.
        
        Returns:
        Complete ASGI scope with request metadata
        """

Response Classes

Classes for creating HTTP responses with control over content, headers, status codes, and media types.

class Response:
    """
    Basic HTTP response class.
    """
    def __init__(
        self,
        content: Any = None,
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: Optional[str] = None,
        background: Optional[BackgroundTask] = None
    ):
        """
        Create HTTP response.
        
        Parameters:
        - content: Response content (string, bytes, or iterable)
        - status_code: HTTP status code
        - headers: Additional response headers
        - media_type: Content-Type header value
        - background: Background task to run after response
        """

class JSONResponse(Response):
    """
    JSON response with automatic serialization.
    """
    def __init__(
        self,
        content: Any = None,
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: str = "application/json",
        background: Optional[BackgroundTask] = None
    ):
        """
        Create JSON response with automatic serialization.
        
        Parameters:
        - content: Python object to serialize as JSON
        - status_code: HTTP status code
        - headers: Additional response headers
        - media_type: Content-Type (defaults to application/json)
        - background: Background task to run after response
        """

class HTMLResponse(Response):
    """
    HTML response for serving HTML content.
    """
    def __init__(
        self,
        content: str = "",
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: str = "text/html",
        background: Optional[BackgroundTask] = None
    ):
        """
        Create HTML response.
        
        Parameters:
        - content: HTML content as string
        - status_code: HTTP status code
        - headers: Additional response headers  
        - media_type: Content-Type (defaults to text/html)
        - background: Background task to run after response
        """

class PlainTextResponse(Response):
    """
    Plain text response.
    """
    def __init__(
        self,
        content: str = "",
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: str = "text/plain",
        background: Optional[BackgroundTask] = None
    ):
        """
        Create plain text response.
        
        Parameters:
        - content: Plain text content
        - status_code: HTTP status code
        - headers: Additional response headers
        - media_type: Content-Type (defaults to text/plain)
        - background: Background task to run after response
        """

class RedirectResponse(Response):
    """
    HTTP redirect response.
    """
    def __init__(
        self,
        url: Union[str, URL],
        status_code: int = 307,
        headers: Optional[Dict[str, str]] = None,
        background: Optional[BackgroundTask] = None
    ):
        """
        Create redirect response.
        
        Parameters:
        - url: Redirect target URL
        - status_code: HTTP redirect status code (301, 302, 307, 308)
        - headers: Additional response headers
        - background: Background task to run after response
        """

class StreamingResponse(Response):
    """
    Streaming response for large or generated content.
    """
    def __init__(
        self,
        content: Union[Iterable[Union[str, bytes]], AsyncIterable[Union[str, bytes]]],
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: Optional[str] = None,
        background: Optional[BackgroundTask] = None
    ):
        """
        Create streaming response.
        
        Parameters:
        - content: Iterable or async iterable of content chunks
        - status_code: HTTP status code
        - headers: Additional response headers
        - media_type: Content-Type header value
        - background: Background task to run after response
        """

class FileResponse(Response):
    """
    File download response.
    """
    def __init__(
        self,
        path: Union[str, os.PathLike],
        status_code: int = 200,
        headers: Optional[Dict[str, str]] = None,
        media_type: Optional[str] = None,
        background: Optional[BackgroundTask] = None,
        filename: Optional[str] = None,
        stat_result: Optional[os.stat_result] = None,
        method: Optional[str] = None
    ):
        """
        Create file download response.
        
        Parameters:
        - path: File system path to file
        - status_code: HTTP status code
        - headers: Additional response headers
        - media_type: Content-Type (automatically detected if None)
        - background: Background task to run after response
        - filename: Filename for Content-Disposition header
        - stat_result: Pre-computed file stat result
        - method: HTTP method for conditional response
        """

High-Performance Response Classes

FastAPI-specific response classes optimized for performance.

class UJSONResponse(JSONResponse):
    """
    High-performance JSON response using ujson library.
    
    Note: Requires 'ujson' package to be installed
    Provides faster JSON serialization than standard library
    """

class ORJSONResponse(JSONResponse):
    """
    High-performance JSON response using orjson library.
    
    Note: Requires 'orjson' package to be installed
    Provides fastest JSON serialization with additional features
    """

Data Structure Classes

Classes for structured access to request data components.

class Headers:
    """
    Case-insensitive HTTP headers mapping.
    
    Supports:
    - Case-insensitive key access
    - Multiple values per header
    - Standard dict-like interface
    """

class QueryParams:
    """
    URL query parameters mapping.
    
    Supports:
    - Multiple values per parameter
    - Automatic type conversion
    - Standard dict-like interface
    """

class FormData:
    """
    Form data container for form submissions.
    
    Supports:
    - Form fields as key-value pairs
    - File uploads as UploadFile objects
    - Multiple values per field name
    """

class URL:
    """
    URL manipulation and component access.
    
    Attributes:
    - scheme: URL scheme (http, https)
    - hostname: Hostname or IP address
    - port: Port number
    - path: URL path
    - query: Query string
    - fragment: URL fragment
    """

class Address:
    """
    Network address representation.
    
    Attributes:
    - host: IP address or hostname
    - port: Port number
    """

Usage Examples

Accessing Request Information

from fastapi import FastAPI, Request
from typing import Dict, Any

app = FastAPI()

@app.get("/request-info")
async def get_request_info(request: Request):
    return {
        "method": request.method,
        "url": str(request.url),
        "base_url": str(request.base_url),
        "headers": dict(request.headers),
        "query_params": dict(request.query_params),
        "path_params": request.path_params,
        "cookies": request.cookies,
        "client_host": request.client.host if request.client else None,
        "user_agent": request.headers.get("user-agent")
    }

@app.post("/inspect-body")
async def inspect_body(request: Request):
    # Access raw body
    body = await request.body()
    
    # Try to parse as JSON
    try:
        json_data = await request.json()
    except:
        json_data = None
    
    return {
        "body_length": len(body),
        "body_preview": body[:100].decode("utf-8", errors="ignore"),
        "json_data": json_data,
        "content_type": request.headers.get("content-type")
    }

Custom Response Classes

from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse, PlainTextResponse, RedirectResponse

app = FastAPI()

@app.get("/html", response_class=HTMLResponse)
async def get_html():
    html_content = """
    <html>
        <head>
            <title>FastAPI HTML Response</title>
        </head>
        <body>
            <h1>Hello HTML!</h1>
            <p>This is an HTML response from FastAPI.</p>
        </body>
    </html>
    """
    return html_content

@app.get("/plain-text", response_class=PlainTextResponse)
async def get_plain_text():
    return "This is plain text response"

@app.get("/custom-json")
async def get_custom_json():
    return JSONResponse(
        content={"message": "Custom JSON response"},
        status_code=201,
        headers={"X-Custom-Header": "custom-value"}
    )

@app.get("/redirect-example")
async def redirect_example():
    return RedirectResponse(url="/new-location", status_code=302)

Streaming Responses

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import io
import json
import time

app = FastAPI()

@app.get("/stream-data")
async def stream_data():
    def generate_data():
        for i in range(100):
            data = {"item": i, "timestamp": time.time()}
            yield f"data: {json.dumps(data)}\n"
            time.sleep(0.1)  # Simulate processing delay
    
    return StreamingResponse(
        generate_data(),
        media_type="text/plain",
        headers={"X-Stream-Type": "data-stream"}
    )

@app.get("/stream-csv")
async def stream_csv():
    def generate_csv():
        yield "id,name,value\n"
        for i in range(1000):
            yield f"{i},item_{i},{i*10}\n"
    
    return StreamingResponse(
        generate_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=data.csv"}
    )

@app.get("/stream-async")
async def stream_async():
    async def generate_async_data():
        for i in range(50):
            data = f"Async chunk {i}\n"
            yield data.encode()
            # Simulate async operation
            await asyncio.sleep(0.05)
    
    return StreamingResponse(
        generate_async_data(),
        media_type="text/plain"
    )

File Responses

from fastapi import FastAPI, HTTPException
from fastapi.responses import FileResponse
import os

app = FastAPI()

@app.get("/download/{filename}")
async def download_file(filename: str):
    file_path = f"/path/to/files/{filename}"
    
    if not os.path.exists(file_path):
        raise HTTPException(status_code=404, detail="File not found")
    
    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream"
    )

@app.get("/download-with-name/{file_id}")
async def download_with_custom_name(file_id: int):
    # Look up file by ID
    file_info = get_file_info(file_id)  # Your database lookup
    
    return FileResponse(
        path=file_info["path"],
        filename=f"report_{file_id}.pdf",
        media_type="application/pdf",
        headers={"X-File-ID": str(file_id)}
    )

Form Data Handling

from fastapi import FastAPI, Request, Form, File, UploadFile
from fastapi.responses import HTMLResponse

app = FastAPI()

@app.get("/form", response_class=HTMLResponse)
async def show_form():
    return """
    <html>
        <body>
            <form action="/submit-form" method="post" enctype="multipart/form-data">
                <input type="text" name="username" placeholder="Username">
                <input type="email" name="email" placeholder="Email">
                <input type="file" name="file">
                <input type="submit" value="Submit">
            </form>
        </body>
    </html>
    """

@app.post("/submit-form")
async def submit_form(request: Request):
    # Access form data through request
    form = await request.form()
    
    return {
        "form_data": {
            "username": form.get("username"),
            "email": form.get("email"),
            "file": {
                "filename": form["file"].filename if "file" in form else None,
                "content_type": form["file"].content_type if "file" in form else None
            }
        },
        "all_fields": list(form.keys())
    }

@app.post("/submit-form-params")
async def submit_form_with_params(
    username: str = Form(...),
    email: str = Form(...),
    file: UploadFile = File(...)
):
    # Access form data through parameter injection
    return {
        "username": username,
        "email": email,
        "file": {
            "filename": file.filename,
            "content_type": file.content_type,
            "size": len(await file.read())
        }
    }

Advanced Request Processing

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import json

app = FastAPI()

@app.middleware("http")
async def log_requests(request: Request, call_next):
    # Log request details
    print(f"Request: {request.method} {request.url}")
    print(f"Headers: {dict(request.headers)}")
    
    response = await call_next(request)
    
    print(f"Response: {response.status_code}")
    return response

@app.post("/webhook")
async def handle_webhook(request: Request):
    # Verify content type
    content_type = request.headers.get("content-type")
    if content_type != "application/json":
        raise HTTPException(status_code=400, detail="Content-Type must be application/json")
    
    # Verify webhook signature (example)
    signature = request.headers.get("x-webhook-signature")
    if not signature:
        raise HTTPException(status_code=400, detail="Missing webhook signature")
    
    # Get raw body for signature verification
    body = await request.body()
    
    # Verify signature (simplified example)
    expected_signature = compute_signature(body)  # Your signature logic
    if signature != expected_signature:
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    # Parse JSON payload
    try:
        payload = json.loads(body)
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")
    
    # Process webhook
    result = process_webhook(payload)  # Your processing logic
    
    return JSONResponse(
        content={"status": "processed", "result": result},
        headers={"X-Webhook-ID": payload.get("id", "unknown")}
    )

@app.get("/client-info")
async def get_client_info(request: Request):
    return {
        "client": {
            "host": request.client.host if request.client else None,
            "port": request.client.port if request.client else None
        },
        "forwarded_for": request.headers.get("x-forwarded-for"),
        "real_ip": request.headers.get("x-real-ip"),
        "user_agent": request.headers.get("user-agent"),
        "accept": request.headers.get("accept"),
        "accept_language": request.headers.get("accept-language"),
        "referer": request.headers.get("referer")
    }

Response Customization

from fastapi import FastAPI, Response, status
from fastapi.responses import JSONResponse
from datetime import datetime

app = FastAPI()

@app.get("/custom-headers")
async def custom_headers(response: Response):
    # Modify response directly
    response.headers["X-Custom-Header"] = "custom-value"
    response.headers["X-Timestamp"] = str(datetime.now())
    response.status_code = status.HTTP_200_OK
    
    return {"message": "Response with custom headers"}

@app.post("/conditional-response")
async def conditional_response(data: dict):
    if "error" in data:
        return JSONResponse(
            status_code=400,
            content={"error": data["error"], "timestamp": str(datetime.now())},
            headers={"X-Error-Type": "user-error"}
        )
    
    return JSONResponse(
        status_code=201,
        content={"result": "success", "data": data},
        headers={"X-Result-Type": "success"}
    )

@app.get("/cache-control")
async def cache_control():
    return JSONResponse(
        content={"data": "cacheable content"},
        headers={
            "Cache-Control": "public, max-age=3600",
            "ETag": "unique-etag-value",
            "Last-Modified": "Wed, 21 Oct 2023 07:28:00 GMT"
        }
    )

Install with Tessl CLI