FastAPI #

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True, index=True)
    lname = Column(String)
    fname = Column(String)
    email = Column(String, unique=True, index=True)
    todos = relationship("TODO", back_populates="owner", cascade="all, delete-orphan")

class TODO(Base):
    __tablename__ = "todos"
    id = Column(Integer, primary_key=True, index=True)
    text = Column(String, index=True)
    completed = Column(Boolean, default=False)
    owner_id = Column(Integer, ForeignKey("users.id"))
    owner = relationship("User", back_populates="todos")

from pydantic import BaseModel
from pydantic import EmailStr
class UserBase(BaseModel):
    email: EmailStr

class UserCreate(UserBase):
    lname: str  
    fname: str
    password: str

class TODOCreate(BaseModel):
    text: str
    completed: bool

class TODOUpdate(TODOCreate):
    id: int

@app.post("/api/users", response_model=schemas.User)
def signup(user_data: schemas.UserCreate, db: Session = Depends(get_db)):
"""add new user"""
    user = crud.get_user_by_email(db, user_data.email)
    if user:
    raise HTTPException(status_code=409,
        detail="Email already registered.")
        signedup_user = crud.create_user(db, user_data)
    return signedup_user

@app.post("/api/token", response_model=schemas.Token)
def login_for_access_token(db: Session = Depends(get_db),form_data: OAuth2PasswordRequestForm = Depends()):
"""generate access token for valid credentials"""
    user = authenticate_user(db, form_data.username, form_data.password)
    if not user:
    raise HTTPException(status_code=HTTP_401_UNAUTHORIZED,detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"})
        access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
        access_token = create_access_token(data={"sub": user.email},expires_delta=access_token_expires)
    return {"access_token": access_token, "token_type": "bearer"}

def get_db():
"""provide db session to path operation functions"""
    try:
        db = SessionLocal()
    yield db
    finally:
    db.close()

def get_current_user(db: Session = Depends(get_db),token: str = Depends(oauth2_scheme)):
    return decode_access_token(db, token)
@app.get("/api/me", response_model=schemas.User)
def read_logged_in_user(current_user: models.User = Depends(get_current_user)):
"""return user settings for current user"""
    return current_user

@app.get("/api/mytodos", response_model=List[schemas.TODO])
def get_own_todos(current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
"""return a list of TODOs owned by current user"""
    todos = crud.get_user_todos(db, current_user.id)
    return todos

@app.post("/api/todos", response_model=schemas.TODO)
def add_a_todo(todo_data: schemas.TODOCreate, current_user: models.User = Depends(get_current_user),db: Session = Depends(get_db)):
"""add a TODO"""
    todo = crud.create_meal(db, current_user, meal_data)
    return todo

@app.put("/api/todos/{todo_id}", response_model=schemas.TODO)
def update_a_todo(todo_id: int, todo_data: schemas.TODOUpdate, current_user: models.User = Depends(get_current_user), db: Session = Depends(get_db)):
"""update and return TODO for given id"""
    todo = crud.get_todo(db, todo_id)
    updated_todo = crud.update_todo(db, todo_id, todo_data)
    return updated_todo

@app.delete("/api/todos/{todo_id}")
def delete_a_meal(todo_id: int,current_user: models.User = Depends(get_current_user),db: Session = Depends(get_db)):
"""delete TODO of given id"""
    crud.delete_meal(db, todo_id)
    return {"detail": "TODO Deleted"}

from passlib.context import CryptContext

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

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 authenticate_user(db, email: str, password: str):
    user = crud.get_user_by_email(db, email)
    if not user:
    return False
    if not verify_password(password, user.hashed_password):
    return False
    return user

import jwt
from fastapi.security import OAuth2PasswordBearer

SECRET_KEY = os.environ['SECRET_KEY']
ALGORITHM = os.environ['ALGORITHM']

def create_access_token(*, data: dict, expires_delta: 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

def decode_access_token(db, token):
    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])
        email: str = payload.get("sub")
    if email is None:
    raise credentials_exception
        token_data = schemas.TokenData(email=email)
    except PyJWTError:
    raise credentials_exception
        user = crud.get_user_by_email(db, email=token_data.email)
    if user is None:
    raise credentials_exception
    return user

def create_todo(db: Session, current_user: models.User, todo_data: schemas.TODOCreate):
    todo = models.TODO(text=todo_data.text, completed=todo_data.completed)
    todo.owner = current_user
    db.add(todo)
    db.commit()
    db.refresh(todo)
    return todo

def update_todo(db: Session, todo_data: schemas.TODOUpdate):
    todo = db.query(models.TODO).filter(models.TODO.id == id).first()
    todo.text = todo_data.text
    todo.completed = todo.completed
    db.commit()
    db.refresh(todo)
    return todo

def delete_todo(db: Session, id: int):
    todo = db.query(models.TODO).filter(models.TODO.id == id).first()
    db.delete(todo)
    db.commit()

def get_user_todos(db: Session, userid: int):
return db.query(models.TODO).filter(models.TODO.owner_id == userid).all()

# 確保只有登錄用戶可以創建TODO
from starlette.testclient import TestClient
from .main import app
client = TestClient(app)
def test_unauthenticated_user_cant_create_todos(): todo=dict(text="run a mile", completed=False)
    response = client.post("/api/todos", data=todo)
    assert response.status_code == 401

# 測試檢查登錄端點並在提供有效登錄憑據時生成 JWT
def test_user_can_obtain_auth_token():
    response = client.post("/api/token", data=good_credentials)
    assert response.status_code == 200
    assert 'access_token' in response.json()
    assert 'token_type' in response.json()