| """
|
| Production-style FastAPI Task Manager REST API
|
| Single-file implementation with JWT auth, SQLAlchemy, and SQLite.
|
|
|
| Run with:
|
| uvicorn main:app --reload
|
|
|
| Then visit http://localhost:8000/docs for interactive API docs.
|
| """
|
|
|
| from datetime import datetime, timedelta
|
| from typing import Optional, List
|
|
|
| from fastapi import FastAPI, Depends, HTTPException, status
|
| from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
|
| from jose import JWTError, jwt
|
| from passlib.context import CryptContext
|
| from pydantic import BaseModel, Field, ConfigDict
|
| from sqlalchemy import create_engine, String, Integer, DateTime, ForeignKey, Boolean
|
| from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, sessionmaker, Session
|
| from sqlalchemy.exc import IntegrityError
|
|
|
|
|
|
|
|
|
| SECRET_KEY = "change-this-in-production-use-env-var"
|
| ALGORITHM = "HS256"
|
| ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24
|
| DATABASE_URL = "sqlite:///./tasks.db"
|
|
|
|
|
|
|
|
|
| engine = create_engine(
|
| DATABASE_URL,
|
| connect_args={"check_same_thread": False},
|
| )
|
| SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
| class Base(DeclarativeBase):
|
| pass
|
|
|
|
|
|
|
|
|
|
|
| class User(Base):
|
| __tablename__ = "users"
|
|
|
| id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
| username: Mapped[str] = mapped_column(String(50), unique=True, index=True, nullable=False)
|
| email: Mapped[str] = mapped_column(String(120), unique=True, index=True, nullable=False)
|
| hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
|
| created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
|
|
|
| class Task(Base):
|
| __tablename__ = "tasks"
|
|
|
| id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
|
| title: Mapped[str] = mapped_column(String(200), nullable=False)
|
| description: Mapped[Optional[str]] = mapped_column(String(1000), nullable=True)
|
| completed: Mapped[bool] = mapped_column(Boolean, default=False)
|
| owner_id: Mapped[int] = mapped_column(Integer, ForeignKey("users.id"), nullable=False)
|
| created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
|
|
|
|
|
|
|
|
|
|
|
| class UserCreate(BaseModel):
|
| username: str = Field(..., min_length=3, max_length=50)
|
| email: str = Field(..., max_length=120)
|
| password: str = Field(..., min_length=6, max_length=100)
|
|
|
|
|
| class UserOut(BaseModel):
|
| model_config = ConfigDict(from_attributes=True)
|
|
|
| id: int
|
| username: str
|
| email: str
|
| created_at: datetime
|
|
|
|
|
| class TaskCreate(BaseModel):
|
| title: str = Field(..., min_length=1, max_length=200)
|
| description: Optional[str] = Field(None, max_length=1000)
|
| completed: bool = False
|
|
|
|
|
| class TaskUpdate(BaseModel):
|
| title: Optional[str] = Field(None, min_length=1, max_length=200)
|
| description: Optional[str] = Field(None, max_length=1000)
|
| completed: Optional[bool] = None
|
|
|
|
|
| class TaskOut(BaseModel):
|
| model_config = ConfigDict(from_attributes=True)
|
|
|
| id: int
|
| title: str
|
| description: Optional[str]
|
| completed: bool
|
| owner_id: int
|
| created_at: datetime
|
|
|
|
|
| class Token(BaseModel):
|
| access_token: str
|
| token_type: str
|
|
|
|
|
| class TokenData(BaseModel):
|
| username: Optional[str] = None
|
|
|
|
|
|
|
|
|
|
|
| pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
|
| oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")
|
|
|
|
|
| def verify_password(plain_password: str, hashed_password: str) -> bool:
|
| return pwd_context.verify(plain_password, hashed_password)
|
|
|
|
|
| def get_password_hash(password: str) -> str:
|
| return pwd_context.hash(password)
|
|
|
|
|
| def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
|
| to_encode = data.copy()
|
| expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
|
| to_encode.update({"exp": expire})
|
| return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
|
|
|
|
|
|
|
|
|
|
| app = FastAPI(
|
| title="Task Manager API",
|
| description="A production-style FastAPI task manager with JWT auth.",
|
| version="1.0.0",
|
| )
|
|
|
|
|
|
|
|
|
|
|
| def get_db():
|
| db = SessionLocal()
|
| try:
|
| yield db
|
| finally:
|
| db.close()
|
|
|
|
|
|
|
|
|
|
|
| def get_current_user(
|
| token: str = Depends(oauth2_scheme),
|
| db: Session = Depends(get_db),
|
| ) -> User:
|
| 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 = db.query(User).filter(User.username == token_data.username).first()
|
| if user is None:
|
| raise credentials_exception
|
| return user
|
|
|
|
|
|
|
|
|
|
|
| @app.post("/auth/register", response_model=UserOut, status_code=status.HTTP_201_CREATED)
|
| def register_user(user: UserCreate, db: Session = Depends(get_db)):
|
| """Register a new user."""
|
| existing = db.query(User).filter(
|
| (User.username == user.username) | (User.email == user.email)
|
| ).first()
|
| if existing:
|
| raise HTTPException(
|
| status_code=status.HTTP_400_BAD_REQUEST,
|
| detail="Username or email already registered",
|
| )
|
| db_user = User(
|
| username=user.username,
|
| email=user.email,
|
| hashed_password=get_password_hash(user.password),
|
| )
|
| db.add(db_user)
|
| db.commit()
|
| db.refresh(db_user)
|
| return db_user
|
|
|
|
|
| @app.post("/auth/login", response_model=Token)
|
| def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
|
| """Login and receive a JWT access token."""
|
| user = db.query(User).filter(User.username == form_data.username).first()
|
| if not user or not verify_password(form_data.password, user.hashed_password):
|
| raise HTTPException(
|
| status_code=status.HTTP_401_UNAUTHORIZED,
|
| detail="Incorrect username or password",
|
| headers={"WWW-Authenticate": "Bearer"},
|
| )
|
| access_token = create_access_token(
|
| data={"sub": user.username},
|
| expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
|
| )
|
| return {"access_token": access_token, "token_type": "bearer"}
|
|
|
|
|
| @app.get("/users/me", response_model=UserOut)
|
| def read_current_user(current_user: User = Depends(get_current_user)):
|
| """Get current authenticated user's profile."""
|
| return current_user
|
|
|
|
|
|
|
|
|
|
|
| @app.post("/tasks", response_model=TaskOut, status_code=status.HTTP_201_CREATED)
|
| def create_task(
|
| task: TaskCreate,
|
| current_user: User = Depends(get_current_user),
|
| db: Session = Depends(get_db),
|
| ):
|
| """Create a new task owned by the current user."""
|
| db_task = Task(
|
| title=task.title,
|
| description=task.description,
|
| completed=task.completed,
|
| owner_id=current_user.id,
|
| )
|
| db.add(db_task)
|
| db.commit()
|
| db.refresh(db_task)
|
| return db_task
|
|
|
|
|
| @app.get("/tasks", response_model=List[TaskOut])
|
| def list_tasks(
|
| current_user: User = Depends(get_current_user),
|
| db: Session = Depends(get_db),
|
| ):
|
| """List all tasks owned by the current user."""
|
| return db.query(Task).filter(Task.owner_id == current_user.id).all()
|
|
|
|
|
| @app.get("/tasks/{task_id}", response_model=TaskOut)
|
| def get_task(
|
| task_id: int,
|
| current_user: User = Depends(get_current_user),
|
| db: Session = Depends(get_db),
|
| ):
|
| """Get a single task by ID (owner-only)."""
|
| task = db.query(Task).filter(Task.id == task_id).first()
|
| if not task:
|
| raise HTTPException(status_code=404, detail="Task not found")
|
| if task.owner_id != current_user.id:
|
| raise HTTPException(status_code=403, detail="Not authorized to view this task")
|
| return task
|
|
|
|
|
| @app.put("/tasks/{task_id}", response_model=TaskOut)
|
| def update_task(
|
| task_id: int,
|
| task_update: TaskUpdate,
|
| current_user: User = Depends(get_current_user),
|
| db: Session = Depends(get_db),
|
| ):
|
| """Update an existing task (owner-only)."""
|
| task = db.query(Task).filter(Task.id == task_id).first()
|
| if not task:
|
| raise HTTPException(status_code=404, detail="Task not found")
|
| if task.owner_id != current_user.id:
|
| raise HTTPException(status_code=403, detail="Not authorized to edit this task")
|
|
|
| update_data = task_update.model_dump(exclude_unset=True)
|
| for field, value in update_data.items():
|
| setattr(task, field, value)
|
|
|
| db.commit()
|
| db.refresh(task)
|
| return task
|
|
|
|
|
| @app.delete("/tasks/{task_id}", status_code=status.HTTP_204_NO_CONTENT)
|
| def delete_task(
|
| task_id: int,
|
| current_user: User = Depends(get_current_user),
|
| db: Session = Depends(get_db),
|
| ):
|
| """Delete a task (owner-only)."""
|
| task = db.query(Task).filter(Task.id == task_id).first()
|
| if not task:
|
| raise HTTPException(status_code=404, detail="Task not found")
|
| if task.owner_id != current_user.id:
|
| raise HTTPException(status_code=403, detail="Not authorized to delete this task")
|
| db.delete(task)
|
| db.commit()
|
| return None
|
|
|
|
|
|
|
|
|
|
|
| if __name__ == "__main__":
|
| import uvicorn
|
| uvicorn.run(app, host="0.0.0.0", port=8000)
|
|
|
