我有2个问题
raise fastapi.exceptions.FastAPIError( fastapi.exceptions.FastAPIError: Invalid args for response field! Hint: check that <class 'sqlalchemy.ext.asyncio.session.AsyncSession'> is a valid Pydantic field type. If you are using a return type annotation that is not a valid Pydantic field (e.g. Union[Response, dict, None]) you can disable generating the response model from the type annotation with the path operation decorator parameter response_model=None.我查看文档并遵循它 我不知道我应该做什么
raise ResponseValidationError(
fastapi.exceptions.ResponseValidationError: 2 validation errors:
{'type': 'get_attribute_error', 'loc': ('response', 'username'), 'msg': 'Error extracting attribute: DetachedInstanceError: Instance <User at 0x1dc76f1b260> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)', 'input': <models.User object at 0x000001DC76F1B260>, 'ctx': {'error': 'DetachedInstanceError: Instance <User at 0x1dc76f1b260> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)'}}
{'type': 'get_attribute_error', 'loc': ('response', 'hashed_password'), 'msg': 'Error extracting attribute: DetachedInstanceError: Instance <User at 0x1dc76f1b260> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)', 'input': <models.User object at 0x000001DC76F1B260>, 'ctx': {'error': 'DetachedInstanceError: Instance <User at 0x1dc76f1b260> is not bound to a Session; attribute refresh operation cannot proceed (Background on this error at: https://sqlalche.me/e/20/bhk3)'}}
当我重复这些数据时，我看到：
sqlalchemy.exc.IntegrityError: (sqlite3.IntegrityError) UNIQUE constraint failed: users.username
[SQL: INSERT INTO users (username, email, country, hashed_password) VALUES (?, ?, ?, ?)]
[parameters: ('dubina', None, None, '$2b$12$pMfZ.UedzpcDT.8yqX365.bunwSRZf6otVy3Iee2al.XkFjwvgJX.')]
(Background on this error at: https://sqlalche.me/e/20/gkpj)
请帮帮我，我不明白
主要
from database import engine, Base
from database import SessionLocal as session_maker
import requests, httpx
from fastapi import FastAPI, Request, HTTPException, Depends
import database, schemas, crud, models
from starlette import status
from fastapi.security import OAuth2PasswordRequestForm
from typing import Annotated
from datetime import datetime, timedelta, timezone
app = FastAPI()
app.state.engine = engine
app.state.session_maker = session_maker
@app.on_event("startup")
async def on_startup():
async with app.state.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
@app.on_event("shutdown")
async def on_shutdown():
# async with app.state.engine.begin() as conn:
# await conn.run_sync(Base.metadata.drop_all)
await app.state.engine.dispose()
@app.post("/create", response_model=schemas.PUser, status_code=status.HTTP_201_CREATED)
async def create_user(request:Request, user:schemas.PUser):
db = request.app.state.session_maker
async with db.begin() as session:
create = await crud.create_user(session, user)
return create
@app.post("/token")
async def login_for_access_token(
request:Request,
form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
) -> schemas.Token:
db = request.app.state.session_maker
async with db.begin() as session:
user = crud.authenticate_user(session, 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=crud.ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = crud.create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return schemas.Token(access_token=access_token, token_type="bearer")
@app.get("/users/me/items/")
async def read_own_items(
current_user: Annotated[schemas.PUser, Depends(crud.get_current_active_user)],
):
return [{"item_id": "Foo", "owner": current_user.username}]
粗鲁
from sqlalchemy.ext.asyncio import AsyncSession
from passlib.context import CryptContext
import models, schemas
from sqlalchemy import delete, select, update
from fastapi.security import OAuth2PasswordBearer
from typing import Annotated
from fastapi import Depends, HTTPException
from starlette import status
import jwt
from datetime import datetime, timedelta, timezone
from jwt.exceptions import InvalidTokenError
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "9ccbfaecd8529f2c2435c64d69cfe7e6b3a49011a4ab60f26ec9ece31024a84f"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
def get_password_hash(password:str):
return pwd_context.hash(password)
async def create_user(db:AsyncSession, user:schemas.PUser):
password = get_password_hash(user.hashed_password)
db_user = models.User(username = user.username, hashed_password = password)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
async def get_user(db:AsyncSession, username: str):
return await db.scalar(select(models.User).where(models.User.username == username))
async def fake_token(db:AsyncSession,token):
user = await get_user(db, token)
return user
async def get_current_user(db:AsyncSession,token: Annotated[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 = schemas.TokenData(username=username)
except InvalidTokenError:
raise credentials_exception
user = get_user(db, username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(current_user:Annotated[schemas.User, Depends(get_current_user)]):
if current_user.disabled:
raise HTTPException(status_code=400, detail='Inactive user')
return current_user
async def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
async def authenticate_user(db:AsyncSession, username:str, password:str):
user = await get_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: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.now(timezone.utc) + expires_delta
else:
expire = datetime.now(timezone.utc) + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
模式
from typing import Union
from pydantic import BaseModel
class User(BaseModel):
username: str
class Config:
orm_mode=True
class PUser(User):
hashed_password: str
class AUser(User):
email: Union[str, None] = None
country: Union[str, None] = None
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str | None = None
我找不到解决问题的方法。
由于您使用的是异步数据库连接，因此您应该将
awaitcommitrefreshasync def create_user(db:AsyncSession, user:schemas.PUser):
password = get_password_hash(user.hashed_password)
db_user = models.User(username = user.username, hashed_password = password)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user