db: Depends(get_db_session) leads to TypeError (issubclass() arg 1 must be a class) #1128
Description
Dear @tiangolo,
the following thing drives me nuts. I started to develop my first FastAPI based REST API and there is something which I cannot solve since some time.
I implemented user authentication based on your tutorial and I get the following error:
Process SpawnProcess-12:
Traceback (most recent call last):
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/pydantic/validators.py", line 560, in find_validators
if issubclass(type_, val_type):
TypeError: issubclass() arg 1 must be a class
During handling of the above exception, another exception occurred:
Traceback (most recent call last):
File "/usr/local/Cellar/python/3.7.4/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/process.py", line 297, in _bootstrap
self.run()
File "/usr/local/Cellar/python/3.7.4/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/process.py", line 99, in run
self._target(*self._args, **self._kwargs)
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/uvicorn/subprocess.py", line 61, in subprocess_started
target(sockets=sockets)
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/uvicorn/main.py", line 382, in run
loop.run_until_complete(self.serve(sockets=sockets))
File "uvloop/loop.pyx", line 1456, in uvloop.loop.Loop.run_until_complete
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/uvicorn/main.py", line 389, in serve
config.load()
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/uvicorn/config.py", line 288, in load
self.loaded_app = import_from_string(self.app)
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/uvicorn/importer.py", line 20, in import_from_string
module = importlib.import_module(module_str)
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/importlib/__init__.py", line 127, in import_module
return _bootstrap._gcd_import(name[level:], package, level)
File "<frozen importlib._bootstrap>", line 1006, in _gcd_import
File "<frozen importlib._bootstrap>", line 983, in _find_and_load
File "<frozen importlib._bootstrap>", line 967, in _find_and_load_unlocked
File "<frozen importlib._bootstrap>", line 677, in _load_unlocked
File "<frozen importlib._bootstrap_external>", line 728, in exec_module
File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
File "./app/main_2.py", line 155, in <module>
async def read_users_me(current_user: User = Depends(get_current_active_user)):
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/routing.py", line 487, in decorator
callbacks=callbacks,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/routing.py", line 431, in add_api_route
callbacks=callbacks,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/routing.py", line 334, in __init__
self.dependant = get_dependant(path=self.path_format, call=self.endpoint)
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 268, in get_dependant
param=param, path=path, security_scopes=security_scopes
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 107, in get_param_sub_dependant
security_scopes=security_scopes,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 143, in get_sub_dependant
use_cache=depends.use_cache,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 268, in get_dependant
param=param, path=path, security_scopes=security_scopes
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 107, in get_param_sub_dependant
security_scopes=security_scopes,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 143, in get_sub_dependant
use_cache=depends.use_cache,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 277, in get_dependant
param=param, default_field_info=params.Query, param_name=param_name
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/fastapi/dependencies/utils.py", line 374, in get_param_field
field_info=field_info,
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/pydantic/fields.py", line 249, in __init__
self.prepare()
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/pydantic/fields.py", line 335, in prepare
self.populate_validators()
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/pydantic/fields.py", line 468, in populate_validators
*(get_validators() if get_validators else list(find_validators(self.type_, self.model_config))),
File "/Users/michael/src/private/trvmp-api/venv/lib/python3.7/site-packages/pydantic/validators.py", line 569, in find_validators
raise RuntimeError(f'error checking inheritance of {type_!r} (type: {display_as_type(type_)})')
RuntimeError: error checking inheritance of <fastapi.params.Depends object at 0x10ad3a2d0> (type: Depends)
This happens only when get_current_user has the following signature.
async def get_current_user(db: Depends(get_db_session), token: str = Depends(oauth2_scheme)):
...
If I do not inject db via Depends(), everything is fine. But for working with a real database I assume to inject the Session object into get_current_user. Am I wrong here?
To see the whole code, have a look below.
If you uncomment line 111 and comment 112 (definition of get_current_user), everything works fine.
from datetime import datetime, timedelta
import jwt
from fastapi import Depends, FastAPI, HTTPException
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt import PyJWTError
from passlib.context import CryptContext
from pydantic import BaseModel
# to get a string like this run:
# openssl rand -hex 32
from starlette.status import HTTP_401_UNAUTHORIZED
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
DATABASE_URL = os.environ['DATABASE_URL']
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db_session() -> Session:
try:
db = SessionLocal()
yield db
finally:
db.close()
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,
}
}
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str = None
class User(BaseModel):
username: str
email: str = None
full_name: str = None
disabled: bool = None
class UserInDB(User):
hashed_password: str
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
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):
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(token: str = Depends(oauth2_scheme)):
async def get_current_user(db: Depends(get_db_session), token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except PyJWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user: User = Depends(get_current_user)):
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=HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, 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(id: int, current_user: User = Depends(get_current_active_user)):
return [{"item_id": "Foo", "owner": current_user.username}]