Utils Module

The utils module provides core utility functions used throughout FastAPI Shield for dependency resolution, request processing, and signature manipulation.

Overview

This module contains low-level utility functions that power the shield system's integration with FastAPI's dependency injection and request handling. While these functions are primarily for internal use, understanding them can be helpful for advanced use cases.

Module Reference

Utility functions for FastAPI Shield dependency resolution and request processing.

This module provides core utility functions used throughout the FastAPI Shield library for handling dependency injection, request body parsing, signature manipulation, and other internal operations.

generate_unique_id_for_fastapi_shield(dependant, path_format)

Generate a unique identifier for FastAPI Shield dependants.

Creates a unique operation ID by combining the dependant's callable name with the path format, then sanitizing the result to ensure it's valid for use in OpenAPI schemas and internal operations.

Parameters:

Name	Type	Description	Default
dependant	Dependant	The FastAPI Dependant object containing the callable.	required
path_format	str	The raw path format string (e.g., "/users/{user_id}").	required

Returns:

Name	Type	Description
str		A sanitized unique identifier string with non-word characters replaced by underscores.

Examples:

>>> dependant = Dependant(call=lambda: None)
>>> dependant.call.__name__ = "get_user"
>>> generate_unique_id_for_fastapi_shield(dependant, "/users/{user_id}")
"get_user_users__user_id_"

get_body_field_from_dependant(dependant, path_format)

Extract body field information from a FastAPI Dependant.

Analyzes a FastAPI Dependant to determine the appropriate body field configuration and whether body fields should be embedded. This is used for proper request body parsing and validation.

Parameters:

Name	Type	Description	Default
dependant	Dependant	The FastAPI Dependant object to analyze.	required
path_format	str	The raw path format string for generating unique IDs.	required

Returns:

Type	Description
tuple[Optional[ModelField], bool]	tuple[Optional[ModelField], bool]: A tuple containing: - The body field (ModelField) if one exists, None otherwise - Boolean indicating whether body fields should be embedded

Examples:

>>> dependant = get_dependant(path="/users", call=endpoint_func)
>>> body_field, embed = get_body_field_from_dependant(dependant, "/users")
>>> if body_field:
...     print(f"Body field type: {body_field.type_}")

get_body_from_request(request, body_field=None) async

Extract and parse the request body based on content type and field configuration.

Handles various content types including JSON, form data, and raw bytes. Properly manages file uploads and form closures to prevent resource leaks. Provides comprehensive error handling for malformed requests.

Parameters:

Name	Type	Description	Default
request	Request	The FastAPI Request object to extract body from.	required
body_field	Optional[ModelField]	Optional ModelField specifying expected body structure. If None, no body parsing is performed.	None

Returns:

Name	Type	Description
Any		The parsed request body. Type depends on content type: - dict/list for JSON content - FormData for form submissions - bytes for other content types - None if no body field specified

Raises:

Type	Description
RequestValidationError	If JSON parsing fails or body is malformed.
HTTPException	If there's an error parsing the request body.

Examples:

>>> # JSON request
>>> body = await get_body_from_request(request, json_body_field)
>>> print(body)  # {'key': 'value'}

>>> # Form request
>>> body = await get_body_from_request(request, form_body_field)
>>> print(body.get('field_name'))  # 'field_value'

get_path_format_from_request_for_endpoint(request)

Extract the path format from a FastAPI request.

Attempts to retrieve the raw path format (with parameter placeholders) from the request's route. Falls back to the actual URL path if the route doesn't have a path_format attribute.

Parameters:

Name	Type	Description	Default
request	Request	The FastAPI Request object to extract path format from.	required

Returns:

Name	Type	Description
str	str	The path format string (e.g., "/users/{user_id}") or the actual request path if format is unavailable.

Examples:

>>> # For a request to /users/123 with route pattern /users/{user_id}
>>> path_format = get_path_format_from_request_for_endpoint(request)
>>> print(path_format)  # "/users/{user_id}"

>>> # For a request without route pattern
>>> path_format = get_path_format_from_request_for_endpoint(request)
>>> print(path_format)  # "/users/123"

get_solved_dependencies(request, path_format, endpoint, dependency_cache) async

Resolve all dependencies for a FastAPI endpoint.

Performs complete dependency resolution for an endpoint, including parsing the request body, resolving nested dependencies, and handling dependency caching. This is the core function used by shields to access resolved dependency values.

Parameters:

Name	Type	Description	Default
request	Request	The FastAPI Request object containing request data.	required
path_format	str	The raw path format string for the endpoint.	required
endpoint	Callable	The endpoint callable to resolve dependencies for.	required
dependency_cache	dict	Dictionary for caching resolved dependencies to avoid duplicate resolution.	required

Returns:

Name	Type	Description
tuple		A tuple containing: - Solved dependencies object with resolved values and any errors - The parsed request body (if any)

Examples:

>>> solved_deps, body = await get_solved_dependencies(
...     request, "/users/{user_id}", endpoint_func, {}
... )
>>> if not solved_deps.errors:
...     user_id = solved_deps.values.get("user_id")
...     print(f"Resolved user_id: {user_id}")

merge_dedup_seq_params(*seqs_of_params)

Merge multiple sequences of Parameters while removing duplicates.

Combines multiple sequences of inspect.Parameter objects, keeping only the first occurrence of each parameter name. This is used when merging parameters from wrapped functions to avoid duplicate parameters in the final signature.

Parameters:

Name	Type	Description	Default
*seqs_of_params	Sequence[Parameter]	Variable number of Parameter sequences to merge.	()

Yields:

Name	Type	Description
Parameter		Unique parameters in the order they first appear.

Examples:

>>> from inspect import Parameter
>>> seq1 = [Parameter('a', Parameter.POSITIONAL_OR_KEYWORD)]
>>> seq2 = [Parameter('b', Parameter.POSITIONAL_OR_KEYWORD)]
>>> seq3 = [Parameter('a', Parameter.KEYWORD_ONLY)]  # duplicate 'a'
>>> merged = list(merge_dedup_seq_params(seq1, seq2, seq3))
>>> [p.name for p in merged]  # ['a', 'b'] - duplicate 'a' removed

prepend_request_to_signature_params_of_function(function)

Prepend a Request parameter to a function's signature parameters.

Creates a new parameter sequence that starts with a positional-only Request parameter, followed by all the original function's parameters. This is used internally to ensure shield functions have access to the Request object.

Parameters:

Name	Type	Description	Default
function	Callable	The callable to prepend the Request parameter to.	required

Yields:

Name	Type	Description
Parameter		The new Request parameter followed by original parameters.

Examples:

>>> def my_func(user_id: int, name: str): pass
>>> params = list(prepend_request_to_signature_params_of_function(my_func))
>>> [p.name for p in params]  # ['request', 'user_id', 'name']
>>> params[0].annotation  # <class 'fastapi.Request'>

rearrange_params(iter_params)

Rearrange function parameters according to Python's parameter ordering rules.

Sorts parameters to follow Python's required parameter order: 1. POSITIONAL_ONLY parameters 2. Required POSITIONAL_OR_KEYWORD parameters (no default) 3. Optional POSITIONAL_OR_KEYWORD parameters (with default) 4. VAR_POSITIONAL (args) 5. KEYWORD_ONLY parameters 6. VAR_KEYWORD (*kwargs)

This function is highly optimized using alternating buffers and minimal operations for performance when processing large parameter lists.

Parameters:

Name	Type	Description	Default
iter_params	Iterator[Parameter]	Iterator of Parameter objects to rearrange.	required

Yields:

Name	Type	Description
Parameter		Parameters in the correct order according to Python rules.

Examples:

>>> from inspect import Parameter, signature
>>> def func(a, *args, b=1, c, **kwargs, d=2): pass
>>> params = signature(func).parameters.values()
>>> arranged = list(rearrange_params(iter(params)))
>>> [p.name for p in arranged]  # ['a', 'c', 'd', 'b', 'args', 'kwargs']

Note

This function handles the special case where POSITIONAL_OR_KEYWORD parameters are split into required and optional categories for proper ordering.

Function Categories

Dependency Resolution

get_solved_dependencies

The core function for resolving FastAPI dependencies within the shield system.

Usage Example:

from fastapi_shield.utils import get_solved_dependencies

async def my_shield(request: Request) -> dict | None:
    # Resolve dependencies manually if needed
    solved_deps, body = await get_solved_dependencies(
        request=request,
        path_format="/users/{user_id}",
        endpoint=my_endpoint_function,
        dependency_cache={}
    )

    if solved_deps.errors:
        return None  # Block request on dependency errors

    # Use resolved dependencies
    user_id = solved_deps.values.get("user_id")
    return {"resolved_user_id": user_id}

get_body_field_from_dependant

Extract body field configuration from FastAPI dependants for proper request parsing.

get_body_from_request

Parse request bodies with comprehensive error handling for various content types.

Request Processing

get_path_format_from_request_for_endpoint

Extract path format information from FastAPI requests for use in dependency resolution.

Usage Example:

from fastapi_shield.utils import get_path_format_from_request_for_endpoint

@shield
def path_aware_shield(request: Request) -> dict | None:
    path_format = get_path_format_from_request_for_endpoint(request)
    print(f"Endpoint path format: {path_format}")  # e.g., "/users/{user_id}"

    # Use path format for custom logic
    if "/admin/" in path_format:
        return validate_admin_access(request)
    return validate_regular_access(request)

generate_unique_id_for_fastapi_shield

Generate unique identifiers for shield-related operations, particularly useful for OpenAPI schema generation.

Signature Manipulation

rearrange_params

Optimize parameter ordering according to Python's parameter rules for performance and compatibility.

Technical Details: - Handles POSITIONAL_ONLY, POSITIONAL_OR_KEYWORD, VAR_POSITIONAL, KEYWORD_ONLY, and VAR_KEYWORD parameters - Uses optimized alternating buffer algorithm for performance - Properly handles required vs optional POSITIONAL_OR_KEYWORD parameters

merge_dedup_seq_params

Merge multiple parameter sequences while removing duplicates, essential for combining parameters from wrapped functions.

prepend_request_to_signature_params_of_function

Add Request parameter to function signatures for shield integration.

Advanced Usage Examples

Custom Dependency Resolution

from fastapi_shield.utils import get_solved_dependencies, get_body_from_request
from fastapi.dependencies.utils import get_dependant

async def advanced_shield_with_custom_resolution(request: Request) -> dict | None:
    """Shield that manually resolves specific dependencies."""

    # Get dependant for a specific function
    dependant = get_dependant(path="/api/users", call=some_function)

    # Resolve dependencies manually
    solved_deps, body = await get_solved_dependencies(
        request=request,
        path_format="/api/users",
        endpoint=some_function,
        dependency_cache={}
    )

    # Check for validation errors
    if solved_deps.errors:
        logger.error(f"Dependency resolution failed: {solved_deps.errors}")
        return None

    # Access resolved values
    db_session = solved_deps.values.get("db")
    current_user = solved_deps.values.get("current_user")

    # Custom validation logic using resolved dependencies
    if db_session and current_user:
        user_permissions = check_permissions(db_session, current_user.id)
        if "admin" in user_permissions:
            return {"user": current_user, "permissions": user_permissions}

    return None

Request Body Processing

from fastapi_shield.utils import get_body_from_request, get_body_field_from_dependant
from fastapi.dependencies.utils import get_dependant

async def body_validation_shield(request: Request) -> dict | None:
    """Shield that validates request body before endpoint processing."""

    try:
        # For endpoints that expect a request body
        dependant = get_dependant(path="/api/data", call=endpoint_function)
        body_field, embed_fields = get_body_field_from_dependant(dependant, "/api/data")

        if body_field:
            # Parse and validate body
            body = await get_body_from_request(request, body_field)

            # Custom body validation
            if isinstance(body, dict):
                required_fields = ["user_id", "action"]
                if all(field in body for field in required_fields):
                    return {"validated_body": body}

        return None

    except Exception as e:
        logger.error(f"Body validation failed: {e}")
        return None

Signature Analysis

from fastapi_shield.utils import (
    merge_dedup_seq_params,
    rearrange_params,
    prepend_request_to_signature_params_of_function
)
from inspect import signature, Parameter

def analyze_endpoint_signature(endpoint_func):
    """Analyze and optimize endpoint function signatures."""

    # Get original parameters
    original_params = list(signature(endpoint_func).parameters.values())

    # Add Request parameter for shield compatibility
    shield_params = list(prepend_request_to_signature_params_of_function(endpoint_func))

    # Merge with additional parameters if needed
    additional_params = [
        Parameter("shield_data", Parameter.KEYWORD_ONLY, annotation=dict, default=None)
    ]

    merged_params = list(merge_dedup_seq_params(shield_params, additional_params))

    # Rearrange according to Python rules
    optimized_params = list(rearrange_params(iter(merged_params)))

    return {
        "original_count": len(original_params),
        "shield_count": len(shield_params),
        "merged_count": len(merged_params),
        "optimized_count": len(optimized_params),
        "parameter_names": [p.name for p in optimized_params]
    }

# Usage
result = analyze_endpoint_signature(my_endpoint)
print(f"Parameter analysis: {result}")

Performance Monitoring

from fastapi_shield.utils import get_solved_dependencies
import time
import asyncio

async def performance_monitoring_shield(request: Request) -> dict | None:
    """Shield that monitors dependency resolution performance."""

    start_time = time.time()

    try:
        solved_deps, body = await get_solved_dependencies(
            request=request,
            path_format=request.url.path,
            endpoint=request.scope.get("endpoint"),
            dependency_cache={}
        )

        resolution_time = time.time() - start_time

        # Log performance metrics
        logger.info(f"Dependency resolution took {resolution_time:.3f}s")

        if resolution_time > 0.1:  # Slow dependency resolution
            logger.warning(f"Slow dependency resolution: {resolution_time:.3f}s")

        if solved_deps.errors:
            return None

        return {
            "performance": {
                "dependency_resolution_time": resolution_time,
                "dependency_count": len(solved_deps.values)
            }
        }

    except Exception as e:
        logger.error(f"Performance monitoring failed: {e}")
        return None

Integration with FastAPI Internals

Custom Dependant Creation

from fastapi.dependencies.utils import get_dependant
from fastapi_shield.utils import get_body_field_from_dependant

def create_custom_dependant(func, path: str):
    """Create a custom dependant with shield-specific configuration."""

    # Create base dependant
    dependant = get_dependant(path=path, call=func)

    # Get body field configuration
    body_field, embed_fields = get_body_field_from_dependant(dependant, path)

    # Custom modifications
    dependant.shield_enabled = True
    dependant.custom_body_field = body_field

    return dependant

Error Handling Utilities

from fastapi_shield.utils import get_body_from_request
from fastapi.exceptions import RequestValidationError

async def robust_request_parser(request: Request, body_field=None):
    """Robust request parsing with comprehensive error handling."""

    try:
        body = await get_body_from_request(request, body_field)
        return {"success": True, "body": body}

    except RequestValidationError as e:
        return {
            "success": False,
            "error": "validation_error",
            "details": e.errors()
        }

    except Exception as e:
        return {
            "success": False,
            "error": "parsing_error",
            "message": str(e)
        }

Performance Considerations

• Caching: Many utility functions benefit from caching resolved dependencies
• Async Operations: Use async versions when dealing with I/O operations
• Parameter Optimization: The rearrange_params function is highly optimized for performance
• Memory Management: Utility functions handle resource cleanup automatically

Internal Architecture

The utils module serves as the bridge between FastAPI Shield and FastAPI's internal systems:

1. Dependency Resolution: Integrates with FastAPI's dependency injection system
2. Request Processing: Handles various content types and parsing scenarios
3. Signature Management: Optimizes function signatures for performance
4. Error Handling: Provides comprehensive error handling and validation

See Also

• Shield Class - Main shield implementation
• ShieldDepends - Dependency injection integration
• OpenAPI Integration - Schema generation utilities