集成 OAuth2 密码流程

from fastapi import FastAPI
from fastapi.security import OAuth2PasswordBearer

app = FastAPI()

# 'token' is the URL path where clients will log in
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# We will create a @app.post("/token") endpoint next!

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

@app.post("/token")
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends()
):
    # 1. Authenticate user (verify username/password)
    # 2. If valid, create an access token
    # 3. Return the token
    pass # Details coming in the next scene!

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel
from typing import Optional

app = FastAPI()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

# --- Mock User & Token Logic (Simplified for this lesson) ---
class UserInDB(BaseModel):
    username: str
    hashed_password: str
    disabled: Optional[bool] = None

class CurrentUser(BaseModel):
    username: str
    email: Optional[str] = None
    full_name: Optional[str] = None
    disabled: Optional[bool] = None

fake_users_db = {
    "testuser": UserInDB(username="testuser", hashed_password="password123", disabled=False),
    "disableduser": UserInDB(username="disableduser", hashed_password="securepass", disabled=True)
}

def get_user_from_db(username: str):
    user_data = fake_users_db.get(username)
    if user_data:
        return CurrentUser(username=user_data.username, full_name=f"{user_data.username} Full", email=f"{user_data.username}@example.com", disabled=user_data.disabled)
    return None

def authenticate_user(username: str, password: str):
    user_in_db = fake_users_db.get(username)
    if not user_in_db or user_in_db.hashed_password != password: # In real app, use password hashing
        return None
    return get_user_from_db(username)

def create_access_token(data: dict):
    # In a real app, use `jwt.encode` with a secret key (from Lesson 1)
    return f"mock_jwt_for_{data['sub']}"
# --- End Mock Logic ---

@app.post("/token")
async def login_for_access_token(
    form_data: OAuth2PasswordRequestForm = Depends()
):
    user = authenticate_user(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"},
        )
    if user.disabled:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Inactive user"
        )
    access_token = create_access_token(data={"sub": user.username})
    return {"access_token": access_token, "token_type": "bearer"}

# Dependency to get the current user from the token
async def get_current_user(token: str = Depends(oauth2_scheme)):
    # In a real app, this would decode and validate the JWT (from Lesson 1)
    if not token.startswith("mock_jwt_for_"):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials",
            headers={"WWW-Authenticate": "Bearer"},
        )
    username = token.replace("mock_jwt_for_", "")
    user = get_user_from_db(username)
    if user is None or user.disabled:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid authentication credentials or inactive user",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return user

# A Protected Endpoint
@app.get("/users/me/", response_model=CurrentUser)
async def read_users_me(current_user: CurrentUser = Depends(get_current_user)):
    return current_user