task manager assignmnet

task_manager/auth.py

+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+from fastapi.security import OAuth2PasswordRequestForm
+from passlib.context import CryptContext
+import models, schemas
+from database import get_db
+from jose import JWTError, jwt
+from datetime import datetime, timedelta
+
+router = APIRouter()
+
+# Secret key and algorithm (you can generate your own)
+SECRET_KEY = "supersecretkey123"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+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, username: str, password: str):
+    user = db.query(models.User).filter(models.User.username == username).first()
+    if not user or not verify_password(password, user.hashed_password):
+        return False
+    return user
+
+def create_access_token(data: dict, expires_delta: timedelta = None):
+    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)
+
+# @router.post("/login", response_model=schemas.Token)
+# def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
+#     user = authenticate_user(db, form_data.username, form_data.password)
+#     if not user:
+#         raise HTTPException(status_code=400, detail="Incorrect username or password")
+#     access_token = create_access_token(data={"sub": user.username})
+#     return {"access_token": access_token, "token_type": "bearer"}
+
+@router.post("/login", response_model=schemas.Token, tags=["Authentication"], summary="User Login", description="Authenticate a user and return a JWT token.")
+def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
+    user = authenticate_user(db, form_data.username, form_data.password)
+    if not user:
+        raise HTTPException(status_code=400, detail="Incorrect username or password")
+    
+    access_token = create_access_token(data={"sub": user.username})
+    return {"access_token": access_token, "token_type": "bearer"}
\ No newline at end of file

task_manager/crud.py

+from sqlalchemy.orm import Session
+import models, schemas
+from fastapi import HTTPException
+
+def get_user_by_username(db: Session, username: str):
+    return db.query(models.User).filter(models.User.username == username).first()
+
+def create_user(db: Session, user: schemas.UserCreate, hashed_password: str):
+    db_user = models.User(username=user.username, hashed_password=hashed_password)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+def get_todos_by_user(db: Session, user_id: int):
+    return db.query(models.Todo).filter(models.Todo.owner_id == user_id).all()
+
+def create_todo(db: Session, todo: schemas.TodoCreate, user_id: int):
+    db_todo = models.Todo(**todo.dict(), owner_id=user_id)
+    db.add(db_todo)
+    db.commit()
+    db.refresh(db_todo)
+    return db_todo
+
+def update_todo(db: Session, todo_id: int, todo: schemas.TodoCreate):
+    db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()
+    if not db_todo:
+        raise HTTPException(status_code=404, detail="Todo not found")
+    
+    db_todo.title = todo.title
+    db_todo.completed = todo.completed
+    
+    db.commit()
+    db.refresh(db_todo)
+    return db_todo
+
+def delete_todo(db: Session, todo_id: int):
+    db_todo = db.query(models.Todo).filter(models.Todo.id == todo_id).first()
+    if not db_todo:
+        raise HTTPException(status_code=404, detail="Todo not found")
+    
+    db.delete(db_todo)
+    db.commit()
+    return {"detail": "Todo deleted successfully"}
\ No newline at end of file

task_manager/database.py

+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"  # or use your actual DB URL
+
+engine = create_engine(
+    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
+)
+
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()
+
+# Dependency for FastAPI routes
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
\ No newline at end of file

task_manager/main.py

+SECRET_KEY = "supersecretkey123"
+ALGORITHM = "HS256"
+
+from fastapi import FastAPI, Depends, WebSocket, WebSocketDisconnect
+from sqlalchemy.orm import Session  # Import the Session class
+from database import engine, Base, SessionLocal, get_db
+import models, schemas, crud
+from auth import router as auth_router
+from auth import get_password_hash
+from websocket_manager import connect_client, disconnect_client, notify_all
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from fastapi import HTTPException
+
+app = FastAPI()
+Base.metadata.create_all(bind=engine)
+app.include_router(auth_router)
+
+@app.post("/signup", response_model=schemas.User)
+def signup(user: schemas.UserCreate, db: Session = Depends(get_db)):
+    existing_user = crud.get_user_by_username(db, user.username)
+    if existing_user:
+        raise HTTPException(status_code=400, detail="Username already registered")
+    hashed_password = get_password_hash(user.password)
+    return crud.create_user(db, user, hashed_password)
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)):
+    credentials_exception = HTTPException(
+        status_code=401,
+        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
+    except JWTError:
+        raise credentials_exception
+    user = crud.get_user_by_username(db, username)
+    if user is None:
+        raise credentials_exception
+    return user
+
+# @app.get("/")
+# def root():
+#     return {"message": "Welcome to the Task Manager 🚀"}
+
+
+@app.get("/todos/", response_model=list[schemas.Todo])
+def get_user_todos(
+    db: Session = Depends(get_db),
+    current_user: models.User = Depends(get_current_user)
+):
+    return crud.get_todos_by_user(db, user_id=current_user.id)
+
+@app.post("/todos", response_model=schemas.Todo)
+async def create_todo(
+    todo: schemas.TodoCreate,
+    db: Session = Depends(get_db),  # This retrieves the database session
+    current_user: models.User = Depends(get_current_user)  # This retrieves the current user
+):
+    new_todo = crud.create_todo(db=db, todo=todo, user_id=current_user.id)
+    await notify_all({"event": "new_todo", "todo": new_todo})
+    return new_todo
+
+@app.put("/todos/{todo_id}", response_model=schemas.Todo)
+async def update_todo(
+    todo_id: int,
+    todo: schemas.TodoCreate,
+    db: Session = Depends(get_db),
+    current_user: models.User = Depends(get_current_user)
+):
+    updated_todo = crud.update_todo(db=db, todo_id=todo_id, todo=todo)
+    await notify_all({"event": "update_todo", "todo": updated_todo})
+    return updated_todo
+
+@app.delete("/todos/{todo_id}", response_model=dict)
+async def delete_todo(
+    todo_id: int,
+    db: Session = Depends(get_db),
+    current_user: models.User = Depends(get_current_user)
+):
+    return crud.delete_todo(db=db, todo_id=todo_id)
+
+@app.websocket("/ws")
+async def websocket_endpoint(websocket: WebSocket):
+    await connect_client(websocket)
+    try:
+        while True:
+            await websocket.receive_text()  # Keep the connection open
+    except WebSocketDisconnect:
+        await disconnect_client(websocket)
\ No newline at end of file

task_manager/models.py

+from sqlalchemy import Column, Integer, String, Boolean, ForeignKey
+from sqlalchemy.orm import relationship
+from database import Base
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    username = Column(String, unique=True, index=True)
+    hashed_password = Column(String)
+
+    todos = relationship("Todo", back_populates="owner")
+
+class Todo(Base):
+    __tablename__ = "todos"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String)
+    completed = Column(Boolean, default=False)
+    owner_id = Column(Integer, ForeignKey("users.id"))
+
+    owner = relationship("User", back_populates="todos")
\ No newline at end of file

task_manager/requirements.txt

+fastapi
+uvicorn
+sqlalchemy
+databases
+pydantic
+passlib[bcrypt]
+python-jose[cryptography]
+aiosqlite
\ No newline at end of file

task_manager/schemas.py

+from pydantic import BaseModel
+from typing import List, Optional
+
+class TodoBase(BaseModel):
+    title: str
+    completed: Optional[bool] = False
+
+class TodoCreate(TodoBase):
+    pass
+
+class Todo(TodoBase):
+    id: int
+    owner_id: int
+
+    class Config:
+        from_attributes = True
+
+class UserBase(BaseModel):
+    username: str
+
+class UserCreate(UserBase):
+    password: str
+
+class User(UserBase):
+    id: int
+
+    class Config:
+        from_attributes = True
+
+class Token(BaseModel):
+    access_token: str
+    token_type: str
+
+class TodoBase(BaseModel):
+    title: str
+    description: str
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
\ No newline at end of file

task_manager/tasks.py

+from fastapi import APIRouter, Depends, HTTPException
+from models import tasks
+from schemas import TaskCreate
+from database import database
+from websocket_manager import notify_all
+
+router = APIRouter()
+
+@router.post("/tasks/")
+async def create_task(task: TaskCreate, user_id: int = 1):  # demo: hardcoded user
+    query = tasks.insert().values(title=task.title, user_id=user_id)
+    task_id = await database.execute(query)
+    await notify_all({"event": "new_task", "title": task.title})
+    return {"id": task_id, "title": task.title}
\ No newline at end of file

task_manager/websocket_manager.py

+from fastapi import WebSocket
+from typing import List
+
+clients: List[WebSocket] = []
+
+async def connect_client(websocket: WebSocket):
+    await websocket.accept()
+    clients.append(websocket)
+
+async def disconnect_client(websocket: WebSocket):
+    clients.remove(websocket)
+
+async def notify_all(message: dict):
+    for client in clients:
+        await client.send_json(message)
\ No newline at end of file