ShieldDepends Class¶

The ShieldDepends class enables shield-validated data to be injected into FastAPI endpoints as dependencies, providing seamless integration with FastAPI's dependency injection system.

Overview¶

ShieldDepends is a dependency wrapper that integrates shields with FastAPI's dependency injection. It allows endpoints to receive shield-validated data through the standard dependency injection mechanism, making it easy to access authenticated user data, validated permissions, or any other shield-processed information.

Class Reference¶

Bases: Generic[U], Security

A dependency wrapper that integrates shields with FastAPI's dependency injection system.

ShieldDepends allows shield-validated data to be injected into FastAPI endpoints as dependencies. It extends FastAPI's Security class to provide authentication and authorization capabilities while maintaining compatibility with the standard dependency injection system.

The class manages the lifecycle of shielded dependencies: 1. Initially blocked (unblocked=False) to prevent unauthorized access 2. Unblocked after successful shield validation 3. Dependency resolution with shield-provided data 4. Injection into endpoint parameters

Attributes:

Name	Type	Description
`dependency`		The FastAPI dependency callable
`shielded_dependency`		The original dependency function to be shielded
`unblocked`		Flag indicating whether the shield has validated the request
`auto_error`		Whether to automatically raise HTTP errors on validation failure

Examples:

def get_current_user(user_data: dict) -> User:
    return User(**user_data)

@shield
def auth_shield(request: Request) -> Optional[dict]:
    # Validate authentication and return user data
    return {"id": 1, "name": "John"}

@app.get("/profile")
@auth_shield
def get_profile(user: User = ShieldedDepends(get_current_user)):
    return {"user": user.name}

`first_param` `cached` `property` ¶

Get the first parameter of the shielded dependency function.

The first parameter is special because it receives the shield's returned data. If the first parameter has no default value, it's considered required and will receive the shield data. Otherwise, it's treated as optional.

Returns:

Type	Description
`Optional[Parameter]`	Optional[Parameter]: The first parameter if it has no default value, None if the dependency has no parameters or the first parameter has a default value.

`rest_params` `cached` `property` ¶

Get all parameters except the first one (if it has no default).

These parameters will be resolved using FastAPI's standard dependency injection system, while the first parameter (if it exists and has no default) receives the shield's validated data.

Yields:

Name	Type	Description
`Parameter`		All parameters that should be resolved via dependency injection.

`signature` `cached` `property` ¶

Generate the rearranged signature for FastAPI dependency resolution.

Creates a signature containing only the parameters that should be resolved by FastAPI's dependency injection system (excludes the first parameter if it receives shield data).

Returns:

Name	Type	Description
`Signature`	`Signature`	The signature for FastAPI dependency resolution.

`init(shielded_dependency=None, *, auto_error=True, scopes=None, use_cache=True)` ¶

Initialize a new ShieldDepends instance.

Parameters:

Name	Type	Description	Default
`shielded_dependency`	`Optional[U]`	The dependency function to be protected by shields. Can accept data returned by the shield as its first parameter.	`None`
`auto_error`	`bool`	Whether to automatically raise HTTP errors on failure. If False, returns default responses instead.	`True`
`scopes`	`Optional[Sequence[str]]`	OAuth2 scopes required for this dependency (inherited from Security).	`None`
`use_cache`	`bool`	Whether to cache the dependency result (inherited from Security).	`True`

Examples:

# Basic shielded dependency
user_dep = ShieldDepends(get_current_user)

# With custom error handling
user_dep = ShieldDepends(get_current_user, auto_error=False)

# With OAuth2 scopes
admin_dep = ShieldDepends(get_admin_user, scopes=["admin"])

`call(*args, **kwargs)` `async` ¶

Execute the shielded dependency if unblocked.

This method is called by FastAPI's dependency injection system. If the shield has validated the request (unblocked=True), the dependency function is executed with the provided arguments. Otherwise, returns self to indicate the dependency is still blocked.

Parameters:

Name	Type	Description	Default
`*args`		Positional arguments for the dependency function	`()`
`**kwargs`		Keyword arguments for the dependency function	`{}`

Returns:

Name	Type	Description
`Any`		The result of the dependency function if unblocked, or self if still blocked.

`bool()` ¶

Return the unblocked status as a boolean.

Returns:

Name	Type	Description
`bool`		True if the dependency is unblocked, False otherwise.

`resolve_dependencies(request, path_format)` `async` ¶

Resolve the dependencies for this shielded dependency.

Uses FastAPI's dependency resolution system to resolve all the dependency's parameters (except the first one if it receives shield data).

Parameters:

Name	Type	Description	Default
`request`	`Request`	The FastAPI request object	required
`path_format`	`str`	The raw path format string for the endpoint	required

Returns:

Name	Type	Description
`tuple`		Solved dependencies and request body

`_as_unblocked()` `async` ¶

Context manager to temporarily unblock the dependency.

This is used internally during dependency resolution to temporarily allow the dependency to be called. The dependency is automatically re-blocked when the context exits.

Yields:

Name	Type	Description
`None`		The dependency is unblocked for the duration of the context.

Lifecycle¶

The ShieldDepends lifecycle follows these stages:

Initialization - Created as blocked (unblocked=False)
Shield Validation - Shield runs and validates request
Dependency Resolution - Dependencies are resolved using FastAPI's DI
Unblocking - Temporarily unblocked for function execution
Execution - Dependency function called with shield data
Re-blocking - Automatically re-blocked for security

Usage Examples¶

Basic Authentication Dependency¶

from fastapi import FastAPI, Request, Depends
from fastapi_shield import shield, ShieldDepends

app = FastAPI()

@shield
def auth_shield(request: Request) -> dict | None:
    """Validate JWT token and return user data."""
    token = request.headers.get("Authorization", "").replace("Bearer ", "")
    user_data = validate_jwt_token(token)
    if user_data:
        return {"user_id": user_data["sub"], "username": user_data["username"]}
    return None

def get_current_user(user_data: dict) -> dict:
    """Convert shield data to user object."""
    return {
        "id": user_data["user_id"],
        "username": user_data["username"],
        "is_authenticated": True
    }

@app.get("/profile")
@auth_shield
def get_profile(current_user: dict = ShieldDepends(get_current_user)):
    return {"profile": current_user}

Database Integration¶

from sqlalchemy.orm import Session

def get_db() -> Session:
    """Get database session."""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_user_from_db(
    user_data: dict,  # Receives shield data
    db: Session = Depends(get_db)  # Regular dependency
) -> User:
    """Get user object from database using shield data."""
    user_id = user_data["user_id"]
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(404, "User not found")
    return user

@app.get("/profile")
@auth_shield
def get_profile(user: User = ShieldDepends(get_user_from_db)):
    return {"username": user.username, "email": user.email}

Complex Dependencies with Multiple Shields¶

@shield
def auth_shield(request: Request) -> dict | None:
    """Authentication shield."""
    # Auth logic
    return {"user_id": 123, "role": "admin"}

@shield  
def permission_shield(request: Request) -> dict | None:
    """Permission check shield."""
    # Permission logic
    return {"permissions": ["read", "write", "admin"]}

def get_user_with_permissions(
    auth_data: dict,  # From auth_shield
    db: Session = Depends(get_db)
) -> dict:
    """Get user with authentication data."""
    return {"user": get_user(auth_data["user_id"]), "role": auth_data["role"]}

def get_permission_context(
    permission_data: dict,  # From permission_shield
    cache: Redis = Depends(get_redis)
) -> dict:
    """Get permission context."""
    return {"permissions": permission_data["permissions"]}

@app.get("/admin/data")
@auth_shield
@permission_shield  
def get_admin_data(
    user_context: dict = ShieldDepends(get_user_with_permissions),
    perm_context: dict = ShieldDepends(get_permission_context)
):
    return {
        "user": user_context,
        "permissions": perm_context,
        "data": "sensitive admin data"
    }

Optional Shield Data¶

def optional_user_dependency(
    # No first parameter means shield data is optional
    db: Session = Depends(get_db),
    cache: Redis = Depends(get_redis)
) -> dict:
    """Dependency that works with or without shield data."""
    return {"anonymous": True, "db": db, "cache": cache}

@app.get("/public-or-private")
@auth_shield  # May pass or fail
def flexible_endpoint(
    context: dict = ShieldDepends(optional_user_dependency)
):
    if context.get("anonymous"):
        return {"message": "Public access", "user": None}
    else:
        return {"message": "Authenticated access", "user": context.get("user")}

Custom Error Handling¶

def get_user_with_error_handling(
    user_data: dict,
    db: Session = Depends(get_db)
) -> User:
    """Get user with custom error handling."""
    try:
        user_id = user_data["user_id"]
        user = db.query(User).filter(User.id == user_id).first()
        if not user:
            raise HTTPException(404, f"User {user_id} not found")
        return user
    except KeyError:
        raise HTTPException(400, "Invalid user data from shield")

# Use with auto_error=False for custom responses
user_dep = ShieldDepends(get_user_with_error_handling, auto_error=False)

@app.get("/profile")
@auth_shield
def get_profile(user: User = user_dep):
    return {"profile": user.to_dict()}

Integration Patterns¶

With OAuth2 Scopes¶

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def get_user_with_scopes(
    token_data: dict,
    db: Session = Depends(get_db)
) -> User:
    """Get user with OAuth2 scope validation."""
    user = get_user_by_id(db, token_data["user_id"])
    # Scope validation can be done here or in the shield
    return user

# ShieldDepends with OAuth2 scopes
admin_user_dep = ShieldDepends(
    get_user_with_scopes,
    scopes=["admin"]
)

@app.get("/admin/users")
@auth_shield
def get_all_users(admin: User = admin_user_dep):
    return {"users": get_all_users_from_db()}

Caching Integration¶

def get_cached_user(
    user_data: dict,
    cache: Redis = Depends(get_redis),
    db: Session = Depends(get_db)
) -> User:
    """Get user with caching."""
    user_id = user_data["user_id"]

    # Try cache first
    cached_user = cache.get(f"user:{user_id}")
    if cached_user:
        return User.parse_raw(cached_user)

    # Fallback to database
    user = db.query(User).filter(User.id == user_id).first()
    if user:
        cache.setex(f"user:{user_id}", 300, user.json())  # 5 min cache
    return user

# Use caching with ShieldDepends
cached_user_dep = ShieldDepends(get_cached_user, use_cache=True)

Security Considerations¶

Always start blocked: ShieldDepends instances start in a blocked state for security
Automatic re-blocking: After execution, dependencies are automatically re-blocked
Shield validation required: Dependencies only execute after successful shield validation
Error propagation: Validation errors are properly propagated to FastAPI's error handling

Performance Tips¶

Use use_cache=True for expensive dependency calculations
Keep dependency functions lightweight
Use async dependencies for I/O operations
Consider connection pooling for database dependencies

ShieldDepends Class¶

Overview¶

Class Reference¶

first_param cached property ¶

rest_params cached property ¶

__signature__ cached property ¶

__init__(shielded_dependency=None, *, auto_error=True, scopes=None, use_cache=True) ¶

__call__(*args, **kwargs) async ¶

__bool__() ¶

resolve_dependencies(request, path_format) async ¶

_as_unblocked() async ¶

Lifecycle¶

Usage Examples¶

Basic Authentication Dependency¶

Database Integration¶

Complex Dependencies with Multiple Shields¶

Optional Shield Data¶

Custom Error Handling¶

Integration Patterns¶

With OAuth2 Scopes¶

Caching Integration¶

Security Considerations¶

Performance Tips¶

See Also¶

`first_param` `cached` `property` ¶

`rest_params` `cached` `property` ¶

`signature` `cached` `property` ¶

`init(shielded_dependency=None, *, auto_error=True, scopes=None, use_cache=True)` ¶

`call(*args, **kwargs)` `async` ¶

`bool()` ¶

`resolve_dependencies(request, path_format)` `async` ¶

`_as_unblocked()` `async` ¶