OAuth2 scopes

Warning

The current page still doesn’t have a translation for this language.

But you can help translating it: Contributing.

You can use OAuth2 scopes directly with FastAPI, they are integrated to work seamlessly.

This would allow you to have a more fine-grained permission system, following the OAuth2 standard, integrated into your OpenAPI application (and the API docs).

OAuth2 with scopes is the mechanism used by many big authentication providers, like Facebook, Google, GitHub, Microsoft, Twitter, etc. They use it to provide specific permissions to users and applications.

Every time you “log in with” Facebook, Google, GitHub, Microsoft, Twitter, that application is using OAuth2 with scopes.

In this section you will see how to manage authentication and authorization with the same OAuth2 with scopes in your FastAPI application.

Warning

This is a more or less advanced section. If you are just starting, you can skip it.

You don’t necessarily need OAuth2 scopes, and you can handle authentication and authorization however you want.

But OAuth2 with scopes can be nicely integrated into your API (with OpenAPI) and your API docs.

Nevertheless, you still enforce those scopes, or any other security/authorization requirement, however you need, in your code.

In many cases, OAuth2 with scopes can be an overkill.

But if you know you need it, or you are curious, keep reading.

OAuth2 scopes and OpenAPI

The OAuth2 specification defines “scopes” as a list of strings separated by spaces.

The content of each of these strings can have any format, but should not contain spaces.

These scopes represent “permissions”.

In OpenAPI (e.g. the API docs), you can define “security schemes”.

When one of these security schemes uses OAuth2, you can also declare and use scopes.

Each “scope” is just a string (without spaces).

They are normally used to declare specific security permissions, for example:

  • users:read or users:write are common examples.
  • instagram_basic is used by Facebook / Instagram.
  • https://www.googleapis.com/auth/drive is used by Google.

Info

In OAuth2 a “scope” is just a string that declares a specific permission required.

It doesn’t matter if it has other characters like : or if it is a URL.

Those details are implementation specific.

For OAuth2 they are just strings.

Global view

First, let’s quickly see the parts that change from the examples in the main Tutorial - User Guide for OAuth2 with Password (and hashing), Bearer with JWT tokens. Now using OAuth2 scopes:

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Now let’s review those changes step by step.

OAuth2 Security scheme

The first change is that now we are declaring the OAuth2 security scheme with two available scopes, me and items.

The scopes parameter receives a dict with each scope as a key and the description as the value:

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Because we are now declaring those scopes, they will show up in the API docs when you log-in/authorize.

And you will be able to select which scopes you want to give access to: me and items.

This is the same mechanism used when you give permissions while logging in with Facebook, Google, GitHub, etc:

OAuth2 scopes - 图1

JWT token with scopes

Now, modify the token path operation to return the scopes requested.

We are still using the same OAuth2PasswordRequestForm. It includes a property scopes with a list of str, with each scope it received in the request.

And we return the scopes as part of the JWT token.

Danger

For simplicity, here we are just adding the scopes received directly to the token.

But in your application, for security, you should make sure you only add the scopes that the user is actually able to have, or the ones you have predefined.

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Declare scopes in path operations and dependencies

Now we declare that the path operation for /users/me/items/ requires the scope items.

For this, we import and use Security from fastapi.

You can use Security to declare dependencies (just like Depends), but Security also receives a parameter scopes with a list of scopes (strings).

In this case, we pass a dependency function get_current_active_user to Security (the same way we would do with Depends).

But we also pass a list of scopes, in this case with just one scope: items (it could have more).

And the dependency function get_current_active_user can also declare sub-dependencies, not only with Depends but also with Security. Declaring its own sub-dependency function (get_current_user), and more scope requirements.

In this case, it requires the scope me (it could require more than one scope).

Note

You don’t necessarily need to add different scopes in different places.

We are doing it here to demonstrate how FastAPI handles scopes declared at different levels.

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Technical Details

Security is actually a subclass of Depends, and it has just one extra parameter that we’ll see later.

But by using Security instead of Depends, FastAPI will know that it can declare security scopes, use them internally, and document the API with OpenAPI.

But when you import Query, Path, Depends, Security and others from fastapi, those are actually functions that return special classes.

Use SecurityScopes

Now update the dependency get_current_user.

This is the one used by the dependencies above.

Here’s were we are using the same OAuth2 scheme we created before, declaring it as a dependency: oauth2_scheme.

Because this dependency function doesn’t have any scope requirements itself, we can use Depends with oauth2_scheme, we don’t have to use Security when we don’t need to specify security scopes.

We also declare a special parameter of type SecurityScopes, imported from fastapi.security.

This SecurityScopes class is similar to Request (Request was used to get the request object directly).

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Use the scopes

The parameter security_scopes will be of type SecurityScopes.

It will have a property scopes with a list containing all the scopes required by itself and all the dependencies that use this as a sub-dependency. That means, all the “dependants”… this might sound confusing, it is explained again later below.

The security_scopes object (of class SecurityScopes) also provides a scope_str attribute with a single string, containing those scopes separated by spaces (we are going to use it).

We create an HTTPException that we can re-use (raise) later at several points.

In this exception, we include the scopes required (if any) as a string separated by spaces (using scope_str). We put that string containing the scopes in the WWW-Authenticate header (this is part of the spec).

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

  1. from datetime import datetime, timedelta
  2. from typing import Annotated
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: str | None = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: str | None = None
  52.     full_name: str | None = None
  53.     disabled: bool | None = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import Annotated, List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: List[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(
  148.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  149. ):
  150.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  151.     if not user:
  152.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  153.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  154.     access_token = create_access_token(
  155.         data={"sub": user.username, "scopes": form_data.scopes},
  156.         expires_delta=access_token_expires,
  157.     )
  158.     return {"access_token": access_token, "token_type": "bearer"}
  161. @app.get("/users/me/", response_model=User)
  162. async def read_users_me(
  163.     current_user: Annotated[User, Depends(get_current_active_user)]
  164. ):
  165.     return current_user
  168. @app.get("/users/me/items/")
  169. async def read_own_items(
  170.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  171. ):
  172.     return [{"item_id": "Foo", "owner": current_user.username}]
  175. @app.get("/status/")
  176. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  177.     return {"status": "ok"}
  1. from datetime import datetime, timedelta
  2. from typing import List, Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  13. from typing_extensions import Annotated
  15. # to get a string like this run:
  16. # openssl rand -hex 32
  17. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  18. ALGORITHM = "HS256"
  19. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  22. fake_users_db = {
  23.     "johndoe": {
  24.         "username": "johndoe",
  25.         "full_name": "John Doe",
  26.         "email": "johndoe@example.com",
  27.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  28.         "disabled": False,
  29.     },
  30.     "alice": {
  31.         "username": "alice",
  32.         "full_name": "Alice Chains",
  33.         "email": "alicechains@example.com",
  34.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  35.         "disabled": True,
  36.     },
  37. }
  40. class Token(BaseModel):
  41.     access_token: str
  42.     token_type: str
  45. class TokenData(BaseModel):
  46.     username: Union[str, None] = None
  47.     scopes: List[str] = []
  50. class User(BaseModel):
  51.     username: str
  52.     email: Union[str, None] = None
  53.     full_name: Union[str, None] = None
  54.     disabled: Union[bool, None] = None
  57. class UserInDB(User):
  58.     hashed_password: str
  61. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  63. oauth2_scheme = OAuth2PasswordBearer(
  64.     tokenUrl="token",
  65.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  66. )
  68. app = FastAPI()
  71. def verify_password(plain_password, hashed_password):
  72.     return pwd_context.verify(plain_password, hashed_password)
  75. def get_password_hash(password):
  76.     return pwd_context.hash(password)
  79. def get_user(db, username: str):
  80.     if username in db:
  81.         user_dict = db[username]
  82.         return UserInDB(**user_dict)
  85. def authenticate_user(fake_db, username: str, password: str):
  86.     user = get_user(fake_db, username)
  87.     if not user:
  88.         return False
  89.     if not verify_password(password, user.hashed_password):
  90.         return False
  91.     return user
  94. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  95.     to_encode = data.copy()
  96.     if expires_delta:
  97.         expire = datetime.utcnow() + expires_delta
  98.     else:
  99.         expire = datetime.utcnow() + timedelta(minutes=15)
  100.     to_encode.update({"exp": expire})
  101.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  102.     return encoded_jwt
  105. async def get_current_user(
  106.     security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
  107. ):
  108.     if security_scopes.scopes:
  109.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  110.     else:
  111.         authenticate_value = "Bearer"
  112.     credentials_exception = HTTPException(
  113.         status_code=status.HTTP_401_UNAUTHORIZED,
  114.         detail="Could not validate credentials",
  115.         headers={"WWW-Authenticate": authenticate_value},
  116.     )
  117.     try:
  118.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  119.         username: str = payload.get("sub")
  120.         if username is None:
  121.             raise credentials_exception
  122.         token_scopes = payload.get("scopes", [])
  123.         token_data = TokenData(scopes=token_scopes, username=username)
  124.     except (JWTError, ValidationError):
  125.         raise credentials_exception
  126.     user = get_user(fake_users_db, username=token_data.username)
  127.     if user is None:
  128.         raise credentials_exception
  129.     for scope in security_scopes.scopes:
  130.         if scope not in token_data.scopes:
  131.             raise HTTPException(
  132.                 status_code=status.HTTP_401_UNAUTHORIZED,
  133.                 detail="Not enough permissions",
  134.                 headers={"WWW-Authenticate": authenticate_value},
  135.             )
  136.     return user
  139. async def get_current_active_user(
  140.     current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
  141. ):
  142.     if current_user.disabled:
  143.         raise HTTPException(status_code=400, detail="Inactive user")
  144.     return current_user
  147. @app.post("/token", response_model=Token)
  148. async def login_for_access_token(
  149.     form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
  150. ):
  151.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  152.     if not user:
  153.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  154.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  155.     access_token = create_access_token(
  156.         data={"sub": user.username, "scopes": form_data.scopes},
  157.         expires_delta=access_token_expires,
  158.     )
  159.     return {"access_token": access_token, "token_type": "bearer"}
  162. @app.get("/users/me/", response_model=User)
  163. async def read_users_me(
  164.     current_user: Annotated[User, Depends(get_current_active_user)]
  165. ):
  166.     return current_user
  169. @app.get("/users/me/items/")
  170. async def read_own_items(
  171.     current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
  172. ):
  173.     return [{"item_id": "Foo", "owner": current_user.username}]
  176. @app.get("/status/")
  177. async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
  178.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  3. from fastapi import Depends, FastAPI, HTTPException, Security, status
  4. from fastapi.security import (
  5.     OAuth2PasswordBearer,
  6.     OAuth2PasswordRequestForm,
  7.     SecurityScopes,
  8. )
  9. from jose import JWTError, jwt
  10. from passlib.context import CryptContext
  11. from pydantic import BaseModel, ValidationError
  13. # to get a string like this run:
  14. # openssl rand -hex 32
  15. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  16. ALGORITHM = "HS256"
  17. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  20. fake_users_db = {
  21.     "johndoe": {
  22.         "username": "johndoe",
  23.         "full_name": "John Doe",
  24.         "email": "johndoe@example.com",
  25.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  26.         "disabled": False,
  27.     },
  28.     "alice": {
  29.         "username": "alice",
  30.         "full_name": "Alice Chains",
  31.         "email": "alicechains@example.com",
  32.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  33.         "disabled": True,
  34.     },
  35. }
  38. class Token(BaseModel):
  39.     access_token: str
  40.     token_type: str
  43. class TokenData(BaseModel):
  44.     username: str | None = None
  45.     scopes: list[str] = []
  48. class User(BaseModel):
  49.     username: str
  50.     email: str | None = None
  51.     full_name: str | None = None
  52.     disabled: bool | None = None
  55. class UserInDB(User):
  56.     hashed_password: str
  59. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  61. oauth2_scheme = OAuth2PasswordBearer(
  62.     tokenUrl="token",
  63.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  64. )
  66. app = FastAPI()
  69. def verify_password(plain_password, hashed_password):
  70.     return pwd_context.verify(plain_password, hashed_password)
  73. def get_password_hash(password):
  74.     return pwd_context.hash(password)
  77. def get_user(db, username: str):
  78.     if username in db:
  79.         user_dict = db[username]
  80.         return UserInDB(**user_dict)
  83. def authenticate_user(fake_db, username: str, password: str):
  84.     user = get_user(fake_db, username)
  85.     if not user:
  86.         return False
  87.     if not verify_password(password, user.hashed_password):
  88.         return False
  89.     return user
  92. def create_access_token(data: dict, expires_delta: timedelta | None = None):
  93.     to_encode = data.copy()
  94.     if expires_delta:
  95.         expire = datetime.utcnow() + expires_delta
  96.     else:
  97.         expire = datetime.utcnow() + timedelta(minutes=15)
  98.     to_encode.update({"exp": expire})
  99.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  100.     return encoded_jwt
  103. async def get_current_user(
  104.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  105. ):
  106.     if security_scopes.scopes:
  107.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  108.     else:
  109.         authenticate_value = "Bearer"
  110.     credentials_exception = HTTPException(
  111.         status_code=status.HTTP_401_UNAUTHORIZED,
  112.         detail="Could not validate credentials",
  113.         headers={"WWW-Authenticate": authenticate_value},
  114.     )
  115.     try:
  116.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  117.         username: str = payload.get("sub")
  118.         if username is None:
  119.             raise credentials_exception
  120.         token_scopes = payload.get("scopes", [])
  121.         token_data = TokenData(scopes=token_scopes, username=username)
  122.     except (JWTError, ValidationError):
  123.         raise credentials_exception
  124.     user = get_user(fake_users_db, username=token_data.username)
  125.     if user is None:
  126.         raise credentials_exception
  127.     for scope in security_scopes.scopes:
  128.         if scope not in token_data.scopes:
  129.             raise HTTPException(
  130.                 status_code=status.HTTP_401_UNAUTHORIZED,
  131.                 detail="Not enough permissions",
  132.                 headers={"WWW-Authenticate": authenticate_value},
  133.             )
  134.     return user
  137. async def get_current_active_user(
  138.     current_user: User = Security(get_current_user, scopes=["me"])
  139. ):
  140.     if current_user.disabled:
  141.         raise HTTPException(status_code=400, detail="Inactive user")
  142.     return current_user
  145. @app.post("/token", response_model=Token)
  146. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  147.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  148.     if not user:
  149.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  150.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  151.     access_token = create_access_token(
  152.         data={"sub": user.username, "scopes": form_data.scopes},
  153.         expires_delta=access_token_expires,
  154.     )
  155.     return {"access_token": access_token, "token_type": "bearer"}
  158. @app.get("/users/me/", response_model=User)
  159. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  160.     return current_user
  163. @app.get("/users/me/items/")
  164. async def read_own_items(
  165.     current_user: User = Security(get_current_active_user, scopes=["items"])
  166. ):
  167.     return [{"item_id": "Foo", "owner": current_user.username}]
  170. @app.get("/status/")
  171. async def read_system_status(current_user: User = Depends(get_current_user)):
  172.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

  1. from datetime import datetime, timedelta
  2. from typing import Union
  4. from fastapi import Depends, FastAPI, HTTPException, Security, status
  5. from fastapi.security import (
  6.     OAuth2PasswordBearer,
  7.     OAuth2PasswordRequestForm,
  8.     SecurityScopes,
  9. )
  10. from jose import JWTError, jwt
  11. from passlib.context import CryptContext
  12. from pydantic import BaseModel, ValidationError
  14. # to get a string like this run:
  15. # openssl rand -hex 32
  16. SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
  17. ALGORITHM = "HS256"
  18. ACCESS_TOKEN_EXPIRE_MINUTES = 30
  21. fake_users_db = {
  22.     "johndoe": {
  23.         "username": "johndoe",
  24.         "full_name": "John Doe",
  25.         "email": "johndoe@example.com",
  26.         "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
  27.         "disabled": False,
  28.     },
  29.     "alice": {
  30.         "username": "alice",
  31.         "full_name": "Alice Chains",
  32.         "email": "alicechains@example.com",
  33.         "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
  34.         "disabled": True,
  35.     },
  36. }
  39. class Token(BaseModel):
  40.     access_token: str
  41.     token_type: str
  44. class TokenData(BaseModel):
  45.     username: Union[str, None] = None
  46.     scopes: list[str] = []
  49. class User(BaseModel):
  50.     username: str
  51.     email: Union[str, None] = None
  52.     full_name: Union[str, None] = None
  53.     disabled: Union[bool, None] = None
  56. class UserInDB(User):
  57.     hashed_password: str
  60. pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
  62. oauth2_scheme = OAuth2PasswordBearer(
  63.     tokenUrl="token",
  64.     scopes={"me": "Read information about the current user.", "items": "Read items."},
  65. )
  67. app = FastAPI()
  70. def verify_password(plain_password, hashed_password):
  71.     return pwd_context.verify(plain_password, hashed_password)
  74. def get_password_hash(password):
  75.     return pwd_context.hash(password)
  78. def get_user(db, username: str):
  79.     if username in db:
  80.         user_dict = db[username]
  81.         return UserInDB(**user_dict)
  84. def authenticate_user(fake_db, username: str, password: str):
  85.     user = get_user(fake_db, username)
  86.     if not user:
  87.         return False
  88.     if not verify_password(password, user.hashed_password):
  89.         return False
  90.     return user
  93. def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
  94.     to_encode = data.copy()
  95.     if expires_delta:
  96.         expire = datetime.utcnow() + expires_delta
  97.     else:
  98.         expire = datetime.utcnow() + timedelta(minutes=15)
  99.     to_encode.update({"exp": expire})
  100.     encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
  101.     return encoded_jwt
  104. async def get_current_user(
  105.     security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
  106. ):
  107.     if security_scopes.scopes:
  108.         authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
  109.     else:
  110.         authenticate_value = "Bearer"
  111.     credentials_exception = HTTPException(
  112.         status_code=status.HTTP_401_UNAUTHORIZED,
  113.         detail="Could not validate credentials",
  114.         headers={"WWW-Authenticate": authenticate_value},
  115.     )
  116.     try:
  117.         payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
  118.         username: str = payload.get("sub")
  119.         if username is None:
  120.             raise credentials_exception
  121.         token_scopes = payload.get("scopes", [])
  122.         token_data = TokenData(scopes=token_scopes, username=username)
  123.     except (JWTError, ValidationError):
  124.         raise credentials_exception
  125.     user = get_user(fake_users_db, username=token_data.username)
  126.     if user is None:
  127.         raise credentials_exception
  128.     for scope in security_scopes.scopes:
  129.         if scope not in token_data.scopes:
  130.             raise HTTPException(
  131.                 status_code=status.HTTP_401_UNAUTHORIZED,
  132.                 detail="Not enough permissions",
  133.                 headers={"WWW-Authenticate": authenticate_value},
  134.             )
  135.     return user
  138. async def get_current_active_user(
  139.     current_user: User = Security(get_current_user, scopes=["me"])
  140. ):
  141.     if current_user.disabled:
  142.         raise HTTPException(status_code=400, detail="Inactive user")
  143.     return current_user
  146. @app.post("/token", response_model=Token)
  147. async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
  148.     user = authenticate_user(fake_users_db, form_data.username, form_data.password)
  149.     if not user:
  150.         raise HTTPException(status_code=400, detail="Incorrect username or password")
  151.     access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
  152.     access_token = create_access_token(
  153.         data={"sub": user.username, "scopes": form_data.scopes},
  154.         expires_delta=access_token_expires,
  155.     )
  156.     return {"access_token": access_token, "token_type": "bearer"}
  159. @app.get("/users/me/", response_model=User)
  160. async def read_users_me(current_user: User = Depends(get_current_active_user)):
  161.     return current_user
  164. @app.get("/users/me/items/")
  165. async def read_own_items(
  166.     current_user: User = Security(get_current_active_user, scopes=["items"])
  167. ):
  168.     return [{"item_id": "Foo", "owner": current_user.username}]
  171. @app.get("/status/")
  172. async def read_system_status(current_user: User = Depends(get_current_user)):
  173.     return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta
from typing import List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Verify the username and data shape

We verify that we get a username, and extract the scopes.

And then we validate that data with the Pydantic model (catching the ValidationError exception), and if we get an error reading the JWT token or validating the data with Pydantic, we raise the HTTPException we created before.

For that, we update the Pydantic model TokenData with a new property scopes.

By validating the data with Pydantic we can make sure that we have, for example, exactly a list of str with the scopes and a str with the username.

Instead of, for example, a dict, or something else, as it could break the application at some point later, making it a security risk.

We also verify that we have a user with that username, and if not, we raise that same exception we created before.

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

from datetime import datetime, timedelta
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

from datetime import datetime, timedelta
from typing import Annotated, List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

from datetime import datetime, timedelta
from typing import List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError
from typing_extensions import Annotated

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta
from typing import Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta
from typing import List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Verify the scopes

We now verify that all the scopes required, by this dependency and all the dependants (including path operations), are included in the scopes provided in the token received, otherwise raise an HTTPException.

For this, we use security_scopes.scopes, that contains a list with all these scopes as str.

Python 3.10+Python 3.9+Python 3.8+Python 3.10+ non-AnnotatedPython 3.9+ non-AnnotatedPython 3.8+ non-Annotated

from datetime import datetime, timedelta
from typing import Annotated

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

from datetime import datetime, timedelta
from typing import Annotated, List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

from datetime import datetime, timedelta
from typing import List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError
from typing_extensions import Annotated

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: Annotated[str, Depends(oauth2_scheme)]
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: Annotated[User, Security(get_current_user, scopes=["me"])]
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(
    current_user: Annotated[User, Depends(get_current_active_user)]
):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: Annotated[User, Security(get_current_active_user, scopes=["items"])]
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: Annotated[User, Depends(get_current_user)]):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: str | None = None
    full_name: str | None = None
    disabled: bool | None = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: timedelta | None = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta
from typing import Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: list[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Tip

Prefer to use the Annotated version if possible.

from datetime import datetime, timedelta
from typing import List, Union

from fastapi import Depends, FastAPI, HTTPException, Security, status
from fastapi.security import (
    OAuth2PasswordBearer,
    OAuth2PasswordRequestForm,
    SecurityScopes,
)
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel, ValidationError

# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30


fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "johndoe@example.com",
        "hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
        "disabled": False,
    },
    "alice": {
        "username": "alice",
        "full_name": "Alice Chains",
        "email": "alicechains@example.com",
        "hashed_password": "$2b$12$gSvqqUPvlXP2tfVFaWK1Be7DlH.PKZbv5H8KnzzVgXXbVxpva.pFm",
        "disabled": True,
    },
}


class Token(BaseModel):
    access_token: str
    token_type: str


class TokenData(BaseModel):
    username: Union[str, None] = None
    scopes: List[str] = []


class User(BaseModel):
    username: str
    email: Union[str, None] = None
    full_name: Union[str, None] = None
    disabled: Union[bool, None] = None


class UserInDB(User):
    hashed_password: str


pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

oauth2_scheme = OAuth2PasswordBearer(
    tokenUrl="token",
    scopes={"me": "Read information about the current user.", "items": "Read items."},
)

app = FastAPI()


def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)


def get_password_hash(password):
    return pwd_context.hash(password)


def get_user(db, username: str):
    if username in db:
        user_dict = db[username]
        return UserInDB(**user_dict)


def authenticate_user(fake_db, username: str, password: str):
    user = get_user(fake_db, username)
    if not user:
        return False
    if not verify_password(password, user.hashed_password):
        return False
    return user


def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt


async def get_current_user(
    security_scopes: SecurityScopes, token: str = Depends(oauth2_scheme)
):
    if security_scopes.scopes:
        authenticate_value = f'Bearer scope="{security_scopes.scope_str}"'
    else:
        authenticate_value = "Bearer"
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": authenticate_value},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
        token_scopes = payload.get("scopes", [])
        token_data = TokenData(scopes=token_scopes, username=username)
    except (JWTError, ValidationError):
        raise credentials_exception
    user = get_user(fake_users_db, username=token_data.username)
    if user is None:
        raise credentials_exception
    for scope in security_scopes.scopes:
        if scope not in token_data.scopes:
            raise HTTPException(
                status_code=status.HTTP_401_UNAUTHORIZED,
                detail="Not enough permissions",
                headers={"WWW-Authenticate": authenticate_value},
            )
    return user


async def get_current_active_user(
    current_user: User = Security(get_current_user, scopes=["me"])
):
    if current_user.disabled:
        raise HTTPException(status_code=400, detail="Inactive user")
    return current_user


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(fake_users_db, form_data.username, form_data.password)
    if not user:
        raise HTTPException(status_code=400, detail="Incorrect username or password")
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user.username, "scopes": form_data.scopes},
        expires_delta=access_token_expires,
    )
    return {"access_token": access_token, "token_type": "bearer"}


@app.get("/users/me/", response_model=User)
async def read_users_me(current_user: User = Depends(get_current_active_user)):
    return current_user


@app.get("/users/me/items/")
async def read_own_items(
    current_user: User = Security(get_current_active_user, scopes=["items"])
):
    return [{"item_id": "Foo", "owner": current_user.username}]


@app.get("/status/")
async def read_system_status(current_user: User = Depends(get_current_user)):
    return {"status": "ok"}

Dependency tree and scopes

Let’s review again this dependency tree and the scopes.

As the get_current_active_user dependency has as a sub-dependency on get_current_user, the scope "me" declared at get_current_active_user will be included in the list of required scopes in the security_scopes.scopes passed to get_current_user.

The path operation itself also declares a scope, "items", so this will also be in the list of security_scopes.scopes passed to get_current_user.

Here’s how the hierarchy of dependencies and scopes looks like:

  • The path operation read_own_items has:
    • Required scopes ["items"] with the dependency:
    • get_current_active_user:
      • The dependency function get_current_active_user has:
        • Required scopes ["me"] with the dependency:
        • get_current_user:
          • The dependency function get_current_user has:
            • No scopes required by itself.
            • A dependency using oauth2_scheme.
            • A security_scopes parameter of type SecurityScopes:
              • This security_scopes parameter has a property scopes with a list containing all these scopes declared above, so:
                • security_scopes.scopes will contain ["me", "items"] for the path operation read_own_items.
                • security_scopes.scopes will contain ["me"] for the path operation read_users_me, because it is declared in the dependency get_current_active_user.
                • security_scopes.scopes will contain [] (nothing) for the path operation read_system_status, because it didn’t declare any Security with scopes, and its dependency, get_current_user, doesn’t declare any scope either.

Tip

The important and “magic” thing here is that get_current_user will have a different list of scopes to check for each path operation.

All depending on the scopes declared in each path operation and each dependency in the dependency tree for that specific path operation.

More details about SecurityScopes

You can use SecurityScopes at any point, and in multiple places, it doesn’t have to be at the “root” dependency.

It will always have the security scopes declared in the current Security dependencies and all the dependants for that specific path operation and that specific dependency tree.

Because the SecurityScopes will have all the scopes declared by dependants, you can use it to verify that a token has the required scopes in a central dependency function, and then declare different scope requirements in different path operations.

They will be checked independently for each path operation.

Check it

If you open the API docs, you can authenticate and specify which scopes you want to authorize.

OAuth2 scopes - 图2

If you don’t select any scope, you will be “authenticated”, but when you try to access /users/me/ or /users/me/items/ you will get an error saying that you don’t have enough permissions. You will still be able to access /status/.

And if you select the scope me but not the scope items, you will be able to access /users/me/ but not /users/me/items/.

That’s what would happen to a third party application that tried to access one of these path operations with a token provided by a user, depending on how many permissions the user gave the application.

About third party integrations

In this example we are using the OAuth2 “password” flow.

This is appropriate when we are logging in to our own application, probably with our own frontend.

Because we can trust it to receive the username and password, as we control it.

But if you are building an OAuth2 application that others would connect to (i.e., if you are building an authentication provider equivalent to Facebook, Google, GitHub, etc.) you should use one of the other flows.

The most common is the implicit flow.

The most secure is the code flow, but is more complex to implement as it requires more steps. As it is more complex, many providers end up suggesting the implicit flow.

Note

It’s common that each authentication provider names their flows in a different way, to make it part of their brand.

But in the end, they are implementing the same OAuth2 standard.

FastAPI includes utilities for all these OAuth2 authentication flows in fastapi.security.oauth2.

Security in decorator dependencies

The same way you can define a list of Depends in the decorator’s dependencies parameter (as explained in Dependencies in path operation decorators), you could also use Security with scopes there.