python - 尝试登录时出现 TypeError

Question

我正在使用 fastapi 创建一个页面进行身份验证，它会在用户登录时生成 jwt 令牌，并且用户的所有详细信息都存储在数据库中。我正在使用 sqlalchemy 创建数据库。但是当我尝试登录 TypeError 时发生：

TypeError：“表”类型的参数不可迭代”。

这是我的代码：

from datetime import datetime, timedelta
from typing import Optional
import sqlalchemy
import databases
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel

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

DATABASE_URL = "postgresql://postgres:Sumeet@024@127.0.0.1:5432/dbtest"
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
users_data = sqlalchemy.Table(
    "user",
    metadata,
    sqlalchemy.Column("username", sqlalchemy.String),
    sqlalchemy.Column("full_name", sqlalchemy.String),
    sqlalchemy.Column("email", sqlalchemy.String),
    sqlalchemy.Column("password", sqlalchemy.String),
    sqlalchemy.Column("role", sqlalchemy.String)
)

engine = sqlalchemy.create_engine(
    DATABASE_URL
)


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


class TokenData(BaseModel):
    username: Optional[str] = None


class Register(BaseModel):
    username: str
    full_name: str
    email: str
    password: str
    role: str


class User(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[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(user_db, username: str, password: str):
    user = get_user(user_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: Optional[timedelta] = 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(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.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 JWTError:
        raise credentials_exception
    user = get_user(users_data, 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("/register", response_model=User)
async def register(user: Register):
    user_register = users_data.insert().values(
        username=user.username,
        full_name=user.full_name,
        email=user.email,
        password=pwd_context.hash(user.password),
        role=user.role
    )
    await database.execute(user_register)
    return {
        **user.dict()
    }


@app.post("/token", response_model=Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(
        users_data, form_data.username, form_data.password)
    if not user:
        raise HTTPException(
            status_code=status.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"}
    return await database.fetch_one(user)


@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 = Depends(get_current_active_user)):
    return [{"item_id": "Foo", "owner": current_user.username}]


# Running the database
@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()

# TraceBack Call
Traceback (most recent call last):
  File "d:\python\trial\env\lib\site-packages\uvicorn\protocols\http\h11_impl.py", line 394, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "d:\python\trial\env\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 45, in __call__
    return await self.app(scope, receive, send)
  File "d:\python\trial\env\lib\site-packages\fastapi\applications.py", line 179, in __call__  
    await super().__call__(scope, receive, send)
  File "d:\python\trial\env\lib\site-packages\starlette\applications.py", line 111, in __call__    await self.middleware_stack(scope, receive, send)
  File "d:\python\trial\env\lib\site-packages\starlette\middleware\errors.py", line 181, in __call__
    raise exc from None
  File "d:\python\trial\env\lib\site-packages\starlette\middleware\errors.py", line 159, in __call__
    await self.app(scope, receive, _send)
  File "d:\python\trial\env\lib\site-packages\starlette\exceptions.py", line 82, in __call__   
    raise exc from None
  File "d:\python\trial\env\lib\site-packages\starlette\exceptions.py", line 71, in __call__   
    await self.app(scope, receive, sender)
  File "d:\python\trial\env\lib\site-packages\starlette\routing.py", line 566, in __call__     
    await route.handle(scope, receive, send)
  File "d:\python\trial\env\lib\site-packages\starlette\routing.py", line 227, in handle       
    await self.app(scope, receive, send)
  File "d:\python\trial\env\lib\site-packages\starlette\routing.py", line 41, in app
    response = await func(request)
  File "d:\python\trial\env\lib\site-packages\fastapi\routing.py", line 182, in app
    raw_response = await run_endpoint_function(
  File "d:\python\trial\env\lib\site-packages\fastapi\routing.py", line 133, in run_endpoint_function
    return await dependant.call(**values)
  File ".\main1.py", line 129, in login_for_access_token
    user = authenticate_user(
  File ".\main1.py", line 82, in authenticate_user
    user = get_user(fake_db, username)
  File ".\main1.py", line 76, in get_user
    if username in db:
TypeError: argument of type 'Table' is not iterable

标签： pythonsqlalchemy