Fastapi: [QUESTION] Parametrizing a class dependency on a sub-dependency

[gpakosz]

First check

• [x] I used the GitHub search to find a similar issue and didn't find it.
• [x] I searched the FastAPI documentation, with the integrated search.
• [x] I already searched in Google "How to X in FastAPI" and didn't find any information.

Description

When dealing with sub-dependencies, the documentation introduces e.g.

oauth2_scheme = OAuth2PasswordBearer(...)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    ...



md5-ee4e091188ae161000fdf4c635bdaee1



class FixedContentQueryChecker:
    def __init__(self, fixed_content: str):
        self.fixed_content = fixed_content

    def __call__(self, q: str = ""):
        if q:
            return self.fixed_content in q
        return False

checker = FixedContentQueryChecker("bar")

@app.get("/query-checker/")
async def read_query_check(fixed_content_included: bool = Depends(checker)):
    return {"fixed_content_in_query": fixed_content_included}



md5-5d26fdc0c6bb6d154823275c2724a5ef



oauth2_scheme = OAuth2PasswordBearer(...)

get_current_user = CurrentUserFetcher(oauth2_scheme)

@app.get("/foo/")
async def get_foo(current_user: User = Depends(get_current_user)):
  ...

Thank you

[alvarogf97]

Do you want to authenticate an user in your system by using a dependency called get_current_user or creates a class dependency?

[gpakosz]

I want get_current_user to be a callable instance of a class dependency. So that I can parametrize the get_current_user on sub-dependency.

[alvarogf97]

Maybe something like this?

import datetime
import jwt
import pydantic
from fastapi import HTTPException
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from fastapi.security import OAuth2PasswordRequestForm
from starlette import status


TOKEN_AUDIENCE = 'urn:users'
TOKEN_URL = '/auth/login/'
TOKEN_LIFETIME_SECONDS = 1000
TOKEN_ALGORITHM = 'HS256'
TOKEN_SECRET = 'yoursupersecret'


authorization = OAuth2PasswordBearer(TOKEN_URL, auto_error=False)


class User():
  """User data class."""

  id: int
  is_verified: bool
  create_date: datetime.datetime
  password: str
  name: str
  email: pydantic.EmailStr
  birth_date: datetime.date


async def get_current_user(
    token: str = Depends(authorization)
) -> User:
  """Get current user from the request metadata.

  Args:
    token (str): authorization token.

  Raises:
    HTTPException: whether there is no authenticated user.

  Returns:
    models.user.schemas.User: the user who performs the request.
  """
  if not token:
    _unauthorized()
  try:
    data = await decode_token(token, TOKEN_AUDIENCE)
    user_id = data.get('user_id')
    if not user_id:
      _unauthorized()
    # Get your user here from database
    return await User.get(user_id)
  except jwt.PyJWTError:
    _unauthorized()


async def generate_token(
    user: User,
    audience: str = TOKEN_AUDIENCE,
    lifetime: int = TOKEN_LIFETIME_SECONDS
) -> str:
  """Generates authentication token for the given user.

  Args:
    user (models.user.schemas.User): the user the token will be generated for.
    audience (str, optional): token audience. Defaults to st.TOKEN_AUDIENCE.
    lifetime (int, optional): token seconds lifetime. Defaults to
      st.TOKEN_LIFETIME_SECONDS

  Returns:
    str: the user authentication token.
  """
  expire = (
      datetime.datetime.utcnow() +
      datetime.timedelta(seconds=int(lifetime)))
  data = {
      'user_id': user.id,
      'aud': audience,
      'exp': expire
  }
  return jwt.encode(
      data, TOKEN_SECRET, algorithm=TOKEN_ALGORITHM).decode('utf-8')


async def decode_token(token: str, audience: str = TOKEN_AUDIENCE):
  """Decode token.

  Args:
    token (str): the token which will be decoded.
    audience (str, optional): token audience. Defaults to st.TOKEN_AUDIENCE.

  Raises:
    jwt.PyJWTError: whether the given token is expired or malformed.

  Returns:
    dict: token decoded data.
  """
  return jwt.decode(
      token, TOKEN_SECRET,
      audience=audience, algorithms=[TOKEN_ALGORITHM])


def verify_and_update_password(
    plain_password: str, hashed_password: str
) -> Tuple[bool, str]:
  """Verify the given password by matching it with the hashed one.

  Args:
    plain_password (str): plain password.
    hashed_password (str): hashed password.

  Returns:
    Tuple[bool, str]: a tuple which contains if the password has been verified
    and the new password hash.
  """
  return pwd_context.verify_and_update(plain_password, hashed_password)


async def authenticate(
    credentials: OAuth2PasswordRequestForm
) -> User:
  """Authenticates a user with the given credentials.

  Args:
    credentials (OAuth2PasswordRequestForm): user credentials.

  Returns:
    schemas.User: the user who match with the given credentials.
  """
  user = await User.get_by_email(credentials.username)

  verified, updated_password_hash = verify_and_update_password(
      credentials.password, user.password)

  if not verified:
    raise Exception('User not verified')

  # Update password hash to a more robust one if needed
  if updated_password_hash is not None:
    user.hashed_password = updated_password_hash
    await User.update(user)

  return user


def _unauthorized(headers=None):
  """Raise HTTPException with 401 status code.

  Args:
    headers: (dict, optional): additional headers to the response. Defaults to
      None.

  Raises:
    HTTPException: elevate handler unauthorized exception.
  """
  detail = 'Not authenticated'
  raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=detail)

Take care with User get get_by_email update, the are operations examples and are not implemented in the example code.

[gpakosz]

@alvarogf97 I'm asking about sub-dependencies when writing dependencies as classes. If you read my question carefully, you'll realize the excerpt you pasted uses sub-dependencies when writing dependencies as free functions.

I know you tried to help but all in all you're just adding noise my question 😐

[gpakosz]

@tiangolo Any idea please?

[tiangolo]

Sure, you can do that, just pass the subdependencies in the __call__().

E.g. copying your fragments, something like:

oauth2_scheme = OAuth2PasswordBearer(...)

class ParameterizedGetUser:
    def __init__(self, some_config: str):
        self.some_config = some_config

    def __call__(self, token: str = Depends(oauth2_scheme)):
        if self.some_config == "something":
            return token
        return False


@app.get("/items/")
async def read_items(user: User = Depends(ParameterizedGetUser(some_config="Foo"))):
    return {"message": "Hello World"}

:warning: I'm just copying your snippets and combining them, you'll have to adapt it and check it with your code.

[gpakosz]

But can the oauth2_scheme instance be passed to ParameterizedGetUser's constructor?

[transfluxus]

I have the same issue for endpoints where I have a sub-dependency that depend on an authorization dependency swagger doesnt seem to get that it need authorization (the lock icon is not shown) and swagger doesn send the token (Bearer in my case).
... edit: in fact, if I have both dependencies it doesn't work. Only when the auth dependency is present it works (even tho, when I have both the lock symbol is shown, but the token not sent)

[tiangolo]

But can the oauth2_scheme instance be passed to ParameterizedGetUser's constructor?

@gpakosz I'm not sure I understand what do you mean with that :thinking:

@transfluxus please create a minimal self-contained example showing the problem, that way it's easier to help.

[transfluxus]

@tiangolo damn in my minimal self-contained example its working... :smiley:

[tiangolo]

@transfluxus Hehe that happens...

Comparing your self-contained example with your complete app will probably help you to find the bug/problem. It could be somewhere in your app's code, or even maybe in FastAPI itself. :shrug:

[gpakosz]

@tiangolo I would like write

oauth2_scheme_variant1 = ...

oauth2_scheme_variant2 = ...

@app.get("/foo/")
async def get_foo(user: User = Depends(ParameterizedGetUser(oauth2_scheme_variant1)):
    return {"message": "foo"}

@app.get("/bar/")
async def get_bar(user: User = Depends(ParameterizedGetUser(oauth2_scheme_variant2))):
    return {"message": "bar"}

So as to write the ParameterizedGetUser once, and inject a sub-dependency via the constructor.

[gpakosz]

@tiangolo in your suggestion:

def __call__(self, token: str = Depends(oauth2_scheme)):

You're hardcoding the dependency in the def __call__ definition.
In other words, I'm asking whether there's a way to make the dependency a member variable of the class that defines the __call__ function

[tiangolo]

Hmm, you can probably do something like:

from typing import Callable
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(...)

app = FastAPI()


def get_dependency(auth: Callable, some_config: str):
    def dependency(value: str = Depends(auth)):
        print(some_config)
        return value

    return dependency


@app.get("/foo/")
async def get_foo(
    current_user: dict = Depends(get_dependency(oauth2_scheme, some_config="bar"))
):
    return "Hello World"

[github-actions[bot]]

Assuming the original issue was solved, it will be automatically closed now. But feel free to add more comments or create new issues.