Fastapi: Question regarding protected routes, Depends() and decorators.

[nicolaipre]

Hi.

I am in the process of moving away from using Flask and Sanic for Web APIs for my Python projects, and would like to start using FastAPI instead as it seems more professional and more suitable for larger projects (in my opinion).

So far I like this library better than the two I have used before, except for one thing.

I have been investigating the documentation of FastAPI regarding protected routes, and was wondering if it is possible to protect a route without passing Depends() as a parameter to the protected function? No offense, but in my personal opinion this makes the syntax look dirty compared to using decorators, or protecting routes by context group such as in Laravel (PHP).

My question is therefore, are there any other possible ways to protect routes somewhat like the example below:

USER = Depends(get_user(fake_users_db, username=username))

@app.get("/profile")
async def get_profile(current_user=USER):
    return current_user.username

Or is it possible to create a decorator for Depends() that can be used instead (similar to what Flask has)?

The main thing is that I do not want to pass Depends() as a parameter like this:

@app.get("/login_basic")
async def login_basic(auth: BasicAuth = Depends(basic_auth)):
    ...

I know this is not an issue and is intended to be like this in the project, but this question is asked out of personal preference of code styling/formatting.

Edit:
Additional question since I mentioned Laravels way of handling routes. Would it be possible to create something similar to this?

Route::middleware('auth:api', 'throttle:20,1')->group(function() {
    Route::get('/usage',   'AbcController@usage')->name('usage-route');
    Route::get('/search/{searchQuery}', 'SearchController@search_method')->name('search-route');
});

from AbcController import Usage
from SearchController import Search

app.add_route_group('authRoutes', [
    app.add_route("/usage", Usage())
    app.add_route("/search/{searchQuery}", Search(...))
])

[Kludex]

You can create the decorator, just use functools.wraps to not lose the function signature.

And you can use the first approach as well, but not in the way you have written, unless get_user(...) returns a callable.
If does, then you can do the way you want.

[mRokita]

I would recommend using the dependencies argument.

@app.get(
    '/restricted',
    dependencies=[Depends(basic_auth)],
)
def restricted():
   ...

[nicolaipre]

Thank you for the responses.

I managed to hack something together somewhat the way I wanted by using app.include_router() and router.add_api_route().

Is it possible to add middleware for a certain APIRouter() instead of FastAPI() app? I need to find a way to protect routes added with router.add_api_route().

[nicolaipre]

I got it working exactly as I wanted. For anyone else looking for a similar structure/setup, see the implementation below.

The issue can be closed.

---

web.py

#!/usr/bin/env python3
# -*- coding: latin-1 -*-

from fastapi import APIRouter, Depends
from app.Http.Controllers.TestController import TestController
from app.Providers.HTTPHeaderAuthProvider import HTTPHeaderAuthentication
PROTECTED = Depends(HTTPHeaderAuthentication(scopes=['valid_role']))

router = APIRouter()
router.add_api_route(methods=['GET'],   path='/',                      endpoint=TestController.home,     name="Home",      dependencies=[]         )
router.add_api_route(methods=['GET'],   path='/profile/{authuser}',    endpoint=TestController.profile,  name="Profile",   dependencies=[PROTECTED])

---

TestController.py

#!/usr/bin/env python3
# -*- coding: latin-1 -*-

from pprint import pprint as print
from fastapi import Request, Response
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse

class TestController:
    def __init__(self):
        pass

    @classmethod
    def test(self, request: Request, response: Response):
        json_content = jsonable_encoder(request.path_params)
        return JSONResponse(content=json_content, media_type="application/json")

    @classmethod
    def home(self, request: Request, response: Response):
        json_content = jsonable_encoder({'message': 'This is the content of home (./)'})
        return JSONResponse(content=json_content, media_type="application/json")

    @classmethod
    def profile(self, request: Request, response: Response):
        json_content = jsonable_encoder({'message': 'This is the content of /profile.'})
        return JSONResponse(content=json_content, media_type="application/json")

---

HTTPHeaderAuthProvider.py

#!/usr/bin/env python3
# -*- coding: latin-1 -*-

# ripped: https://github.com/tiangolo/fastapi/issues/104

import uuid
from fastapi import Header
from typing import Set, List
from starlette.requests import Request
from starlette.exceptions import HTTPException
from starlette.status import HTTP_401_UNAUTHORIZED, HTTP_403_FORBIDDEN
from pydantic import BaseModel, Schema

class User(BaseModel):
    id: uuid.UUID = Schema(
        default=...,
        description='Identifier for this User'
    )
    username: str = Schema(
        default=...,
        description='Username for this User'
    )
    roles: Set[str] = Schema(
        default=set(),
        description='Roles assigned to this User'
    )

# mock repository lookup
_db = dict(
    validUser=User(
        id=uuid.uuid4(),
        username='validUser',
        roles=set(['valid_role'])
    ),
    badUser=User(
        id=uuid.uuid4(),
        username='badUser',
        roles=set()
    )
)

# define header authentication
class HTTPHeaderAuthentication:
    def __init__(self, *, scopes: List[str]):
        self.scopes = set(scopes)

    async def __call__(self, request: Request, authuser: str = Header(None)) -> User:
        user = self.locate_user(username=authuser)
        if not user:
            raise HTTPException(
                status_code=HTTP_403_FORBIDDEN, detail="Not authenticated"
            )
        if not self.has_required_scope(user.roles):
            raise HTTPException(
                status_code=HTTP_401_UNAUTHORIZED, detail=f"{user.username} is not authorized to access this endpoint"
            )
        return user

    def locate_user(self, username: str) -> User:
        """Mock lookup in repository"""
        return _db.get(username, None)

    def has_required_scope(self, user_scopes: Set[str]) -> bool:
        """Verify the user has the desired auth scope"""
        for scope in self.scopes:
            if scope not in user_scopes:
                return False
        return True