Skip to content

Secure Endpoints

Add authentication and permission checks to API endpoints.

Goal

Protect endpoints with JWT authentication and role-based access control.

Prerequisites

Authentication Levels

Level Factory Call Requirement
Public None No authentication
Authenticated _jwt_auth_factory() Valid JWT token
Staff _jwt_auth_factory(require_staff=True) user.is_staff=True
Superuser _jwt_auth_factory(require_superuser=True) user.is_superuser=True

Step-by-Step

1. Inject the Factory

from dataclasses import dataclass

from fastdjango.core.authentication.delivery.fastapi.auth import JWTAuthFactory


@dataclass(kw_only=True)
class ProductController(BaseAsyncController):
    _product_service: ProductService
    _jwt_auth_factory: JWTAuthFactory

    # Auth dependencies are created in __post_init__

2. Create Auth Dependencies in __post_init__

def __post_init__(self) -> None:
    self._jwt_auth = self._jwt_auth_factory()
    self._staff_auth = self._jwt_auth_factory(require_staff=True)
    self._superuser_auth = self._jwt_auth_factory(require_superuser=True)
    super().__post_init__()

3. Apply to Routes

from fastapi import APIRouter, Depends


def register(self, registry: APIRouter) -> None:
    # Public endpoint - no auth
    registry.add_api_route(
        path="/v1/products",
        endpoint=self.list_products,
        methods=["GET"],
    )

    # Authenticated users only
    registry.add_api_route(
        path="/v1/products/favorites",
        endpoint=self.list_favorites,
        methods=["GET"],
        dependencies=[Depends(self._jwt_auth)],
    )

    # Staff only
    registry.add_api_route(
        path="/v1/products",
        endpoint=self.create_product,
        methods=["POST"],
        dependencies=[Depends(self._staff_auth)],
    )

    # Superuser only
    registry.add_api_route(
        path="/v1/products/{product_id}",
        endpoint=self.delete_product,
        methods=["DELETE"],
        dependencies=[Depends(self._superuser_auth)],
    )

4. Access the Authenticated User

Use AuthenticatedRequest to access the user:

from fastdjango.core.authentication.delivery.fastapi.auth import AuthenticatedRequest


async def list_favorites(self, request: AuthenticatedRequest) -> list[ProductSchema]:
    user = request.state.user
    return await self._product_service.list_favorites_for_user(user=user)

Complete Example

# src/fastdjango/core/product/delivery/fastapi/controllers.py
from dataclasses import dataclass
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, status

from fastdjango.core.product.exceptions import ProductNotFoundError
from fastdjango.core.product.services import ProductService
from fastdjango.core.authentication.delivery.fastapi.auth import (
    AuthenticatedRequest,
    JWTAuthFactory,
)
from fastdjango.core.product.delivery.fastapi.schemas import (
    CreateProductRequestSchema,
    ProductSchema,
)
from fastdjango.foundation.delivery.controllers import BaseAsyncController


@dataclass(kw_only=True)
class ProductController(BaseAsyncController):
    _product_service: ProductService
    _jwt_auth_factory: JWTAuthFactory

    def __post_init__(self) -> None:
        self._jwt_auth = self._jwt_auth_factory()
        self._staff_auth = self._jwt_auth_factory(require_staff=True)
        super().__post_init__()

    def register(self, registry: APIRouter) -> None:
        # Public - anyone can view products
        registry.add_api_route(
            path="/v1/products",
            endpoint=self.list_products,
            methods=["GET"],
            response_model=list[ProductSchema],
        )

        # Authenticated - users can view their favorites
        registry.add_api_route(
            path="/v1/products/favorites",
            endpoint=self.list_favorites,
            methods=["GET"],
            response_model=list[ProductSchema],
            dependencies=[Depends(self._jwt_auth)],
        )

        # Staff - only staff can create products
        registry.add_api_route(
            path="/v1/products",
            endpoint=self.create_product,
            methods=["POST"],
            response_model=ProductSchema,
            status_code=status.HTTP_201_CREATED,
            dependencies=[Depends(self._staff_auth)],
        )

    # Public endpoint - no request parameter needed
    async def list_products(self) -> list[ProductSchema]:
        products = await self._product_service.list_products()
        return [
            ProductSchema.model_validate(p, from_attributes=True)
            for p in products
        ]

    # Authenticated endpoint - uses AuthenticatedRequest
    async def list_favorites(
        self,
        request: AuthenticatedRequest,
    ) -> list[ProductSchema]:
        user = request.state.user
        products = await self._product_service.list_favorites(user=user)
        return [
            ProductSchema.model_validate(p, from_attributes=True)
            for p in products
        ]

    # Staff endpoint - uses AuthenticatedRequest
    async def create_product(
        self,
        request: AuthenticatedRequest,
        body: CreateProductRequestSchema,
    ) -> ProductSchema:
        # request.state.user is guaranteed to be staff
        product = await self._product_service.create_product(
            name=body.name,
            price=body.price,
        )
        return ProductSchema.model_validate(product, from_attributes=True)

HTTP Response Codes

Situation Status Code
No token provided 403 Forbidden
Invalid/expired token 401 Unauthorized
Valid token, not staff 403 Forbidden
Valid token, not superuser 403 Forbidden

Rate Limiting

For additional security, add rate limiting:

from fastapi import Depends
from throttled import rate_limiter

from fastdjango.core.shared.delivery.fastapi.throttling import IPThrottlerFactory


@dataclass(kw_only=True)
class AuthController(BaseAsyncController):
    _ip_throttler_factory: IPThrottlerFactory

    def register(self, registry: APIRouter) -> None:
        registry.add_api_route(
            path="/v1/auth/token",
            endpoint=self.issue_token,
            methods=["POST"],
            dependencies=[Depends(self._ip_throttler_factory(quota=rate_limiter.per_min(10)))],
        )

    async def issue_token(
        self,
        request: Request,
        body: IssueTokenRequestSchema,
    ) -> TokenResponseSchema:
        # ... token issuing logic ...

Testing Secured Endpoints

@pytest.mark.django_db(transaction=True)
class TestProductController:
    def test_public_endpoint(
        self,
        test_client_factory: TestClientFactory,
    ) -> None:
        # No auth needed
        with test_client_factory() as client:
            response = client.get("/v1/products")

        assert response.status_code == HTTPStatus.OK

    def test_authenticated_endpoint(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        # auth_for_user auto-generates token
        with test_client_factory(auth_for_user=user) as client:
            response = client.get("/v1/products/favorites")

        assert response.status_code == HTTPStatus.OK

    def test_staff_endpoint_as_regular_user(
        self,
        test_client_factory: TestClientFactory,
        user: User,  # Regular user, not staff
    ) -> None:
        with test_client_factory(auth_for_user=user) as client:
            response = client.post(
                "/v1/products",
                json={"name": "Test", "price": "9.99"},
            )

        assert response.status_code == HTTPStatus.FORBIDDEN

    def test_staff_endpoint_as_staff(
        self,
        test_client_factory: TestClientFactory,
        user_factory: TestUserFactory,
    ) -> None:
        staff_user = user_factory(is_staff=True)

        with test_client_factory(auth_for_user=staff_user) as client:
            response = client.post(
                "/v1/products",
                json={"name": "Test", "price": "9.99"},
            )

        assert response.status_code == HTTPStatus.CREATED

Best Practices

  1. Default to authenticated: Require auth unless explicitly public
  2. Use appropriate level: Don't over-restrict (staff vs superuser)
  3. Check ownership: Even authenticated users should only access their own data
  4. Rate limit sensitive endpoints: Protect login, password reset, etc.
  5. Log security events: Track failed auth attempts