mirror of
https://github.com/a-mayb3/Kanban_clone_backend.git
synced 2026-03-21 18:15:37 +01:00
Compare commits
No commits in common. "7b011fd887690848ffbb5812e7529bf7d46483d6" and "8fb4ba71b951c0b2237704266f4caa44231c9693" have entirely different histories.
7b011fd887
...
8fb4ba71b9
7 changed files with 233 additions and 263 deletions
31
main.py
31
main.py
|
|
@ -10,26 +10,6 @@ from routers.auth import router as auth_router
|
|||
from routers.me import router as me_router
|
||||
from database import init_db
|
||||
|
||||
app_description = """
|
||||
This API serves as the backend for a Kanban-style project management application.
|
||||
It allows users to manage projects, tasks, and user assignments with proper authentication and authorization.
|
||||
|
||||
## Stack
|
||||
- FastAPI
|
||||
- SQLAlchemy
|
||||
- SQLite
|
||||
|
||||
## Features
|
||||
- User Authentication (JWT and Argon2 Password Hashing)
|
||||
- Project Management (Create, Read, Update, Delete)
|
||||
- Task Management within Projects
|
||||
- User Assignments to Projects
|
||||
- CORS Configuration for Frontend Integration
|
||||
|
||||
## Source Code
|
||||
The source code for this API can be found on [GitHub](https://github.com/a-mayb3/Kanban_clone_backend) or [my forgejo instance](https://git.vollex.cc/a-mayb3/Kanban_clone_backend).
|
||||
"""
|
||||
|
||||
global_logger = logging.getLogger()
|
||||
global_logger.setLevel(logging.INFO)
|
||||
logging.basicConfig(
|
||||
|
|
@ -52,12 +32,7 @@ async def lifespan(app: FastAPI):
|
|||
init_db()
|
||||
yield
|
||||
|
||||
app = FastAPI(
|
||||
lifespan=lifespan,
|
||||
license_info={"name": "AGPL-3.0-or-later", "url": "https://www.gnu.org/licenses/agpl-3.0.en.html"},
|
||||
description=app_description,
|
||||
title="Kanban Clone Backend API",
|
||||
)
|
||||
app = FastAPI(lifespan=lifespan)
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
|
|
@ -82,6 +57,10 @@ def ping():
|
|||
def source():
|
||||
return {"url": "https://github.com/a-mayb3/Kanban_clone_backend"}
|
||||
|
||||
## TODO: Add root endpoint that gives basic info about the API
|
||||
## TODO: Add more detailed error handling and logging
|
||||
## TODO: Implement authentication and authorization mechanisms
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
|
|
|
|||
|
|
@ -18,7 +18,7 @@ ACCESS_TOKEN_EXPIRE_MINUTES = 1440 # 24 hours
|
|||
|
||||
def create_access_token(data: dict, expires_delta: timedelta | None = None):
|
||||
"""Create a JWT token"""
|
||||
|
||||
|
||||
to_encode = data.copy()
|
||||
if expires_delta:
|
||||
expire = datetime.now(timezone.utc) + expires_delta
|
||||
|
|
@ -29,63 +29,38 @@ def create_access_token(data: dict, expires_delta: timedelta | None = None):
|
|||
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||||
return encoded_jwt
|
||||
|
||||
def verify_jwt_token(token: str):
|
||||
"""Verify and decode a JWT token"""
|
||||
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id = payload.get("sub")
|
||||
if user_id is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
)
|
||||
return user_id
|
||||
|
||||
except JWTError:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
)
|
||||
|
||||
def get_user_from_jwt(request: Request, db: db_dependency) -> models.User :
|
||||
"""Helper function to check for valid JWT token in cookies"""
|
||||
|
||||
get_token = request.cookies.get("access_token")
|
||||
|
||||
if not get_token or get_token is None:
|
||||
token = request.cookies.get("access_token")
|
||||
if not token:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Not logged in"
|
||||
)
|
||||
try:
|
||||
user_id: str = verify_jwt_token(get_token) ## verifying token validity
|
||||
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id: str = str(payload.get("sub"))
|
||||
if user_id is None:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Not logged in"
|
||||
)
|
||||
db_user = db.query(models.User).filter(models.User.id == int(user_id)).first()
|
||||
if db_user is None:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Could not verify credentials"
|
||||
detail="User not found"
|
||||
)
|
||||
return db_user
|
||||
except HTTPException:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
raise
|
||||
|
||||
def verify_user_password(user_id: int, password: str, db: db_dependency) -> None:
|
||||
"""Verify user's password"""
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if db_user is None:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Could not verify credentials")
|
||||
|
||||
hashed_password = hash(password=password, salt=str(getattr(db_user,"password_salt")), variant="id")
|
||||
if hashed_password != db_user.password_hash:
|
||||
except JWTError:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Could not verify credentials")
|
||||
detail="Could not validate credentials"
|
||||
)
|
||||
|
||||
@router.post("/login")
|
||||
def login(user_data: user_schemas.UserLogin, request: Request, response: Response, db: db_dependency):
|
||||
|
|
@ -103,9 +78,8 @@ def login(user_data: user_schemas.UserLogin, request: Request, response: Respons
|
|||
}
|
||||
}
|
||||
except HTTPException:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
pass # Token invalid or expired, proceed to login
|
||||
|
||||
## check if user exists
|
||||
db_user = db.query(models.User).filter(models.User.email == user_data.email).first()
|
||||
if db_user is None:
|
||||
raise HTTPException(
|
||||
|
|
@ -113,12 +87,16 @@ def login(user_data: user_schemas.UserLogin, request: Request, response: Respons
|
|||
detail="Incorrect email or password"
|
||||
)
|
||||
|
||||
verify_user_password(getattr(db_user, "id"), user_data.password, db)
|
||||
if not verify_user_password(getattr(db_user, "id"), user_data.password, db):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect email or password"
|
||||
)
|
||||
|
||||
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
access_token = create_access_token(
|
||||
data={"sub": str(db_user.id)},
|
||||
expires_delta=access_token_expires
|
||||
data={"sub": str(db_user.id)}, expires_delta=access_token_expires
|
||||
|
||||
)
|
||||
|
||||
# Set JWT in httpOnly cookie
|
||||
|
|
@ -140,3 +118,30 @@ def login(user_data: user_schemas.UserLogin, request: Request, response: Respons
|
|||
}
|
||||
}
|
||||
|
||||
|
||||
def verify_jwt_token(token: str):
|
||||
"""Verify and decode a JWT token"""
|
||||
credentials_exception = HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Could not validate credentials",
|
||||
)
|
||||
try:
|
||||
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
|
||||
user_id = payload.get("sub")
|
||||
if user_id is None:
|
||||
raise credentials_exception
|
||||
return user_id
|
||||
except JWTError:
|
||||
raise credentials_exception
|
||||
|
||||
def verify_user_password(user_id: int, password: str, db: db_dependency) -> bool:
|
||||
"""Verify user's password"""
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if db_user is None:
|
||||
return False
|
||||
|
||||
hashed_password = hash(password=password, salt=str(getattr(db_user,"password_salt")), variant="id")
|
||||
if hashed_password != db_user.password_hash:
|
||||
return False
|
||||
|
||||
return True
|
||||
|
|
@ -35,18 +35,6 @@ def delete_me(request: Request, db: db_dependency):
|
|||
"""Delete current authenticated user"""
|
||||
|
||||
user = auth.get_user_from_jwt(request, db)
|
||||
|
||||
## Remove user from all projects, delete projects with no users left
|
||||
projects = user.projects[:]
|
||||
for project in projects:
|
||||
project.users.remove(user)
|
||||
if len(project.users) == 0:
|
||||
## delete project if no users left
|
||||
tasks = project.tasks[:]
|
||||
for task in tasks:
|
||||
db.delete(task)
|
||||
db.delete(project)
|
||||
|
||||
db.delete(user)
|
||||
db.commit()
|
||||
## Logout user by clearing cookie
|
||||
|
|
|
|||
|
|
@ -1,40 +1,26 @@
|
|||
from fastapi import APIRouter, HTTPException, Depends, Request
|
||||
from typing import List, Annotated
|
||||
|
||||
from httpx import request
|
||||
from jose import JWTError, jwt
|
||||
|
||||
from routers import auth
|
||||
from database import db_dependency
|
||||
|
||||
from schemas.tasks import TaskBase, TaskCreate, TaskUpdate
|
||||
from schemas.projects import ProjectBase, ProjectCreate, ProjectUpdate, ProjectAddUsers, ProjectRemoveUsers
|
||||
from schemas.users import UserBase
|
||||
from schemas.projects_users import ProjectUserBase
|
||||
from schemas.projects_tasks import ProjectTaskBase, ProjectTaskCreate
|
||||
import schemas.tasks as tasks_schemas
|
||||
import schemas.projects as projects_schemas
|
||||
import schemas.users as users_schemas
|
||||
|
||||
from models import Project, Task, User
|
||||
from models import Project
|
||||
from routers.auth import get_user_from_jwt
|
||||
|
||||
def get_project_by_id_for_user(user: UserBase, project_id: int, db: db_dependency) -> ProjectBase:
|
||||
"""Get a project by ID and verify user has access"""
|
||||
db_project = db.query(Project).filter(Project.id == project_id).first()
|
||||
if db_project is None:
|
||||
raise HTTPException(status_code=404, detail="Project not found")
|
||||
if user not in db_project.users:
|
||||
raise HTTPException(status_code=403, detail="Not authorized to access this project")
|
||||
|
||||
return db_project
|
||||
|
||||
def get_task_by_id_for_project(project: ProjectBase, task_id: int, db: db_dependency) -> TaskBase:
|
||||
"""
|
||||
Get a task by ID within a project
|
||||
Supposes the user has already been verified to have access to the project
|
||||
"""
|
||||
db_task = db.query(Task).filter(Task.id == task_id, Task.project_id == getattr(project, "id")).first()
|
||||
if db_task is None:
|
||||
raise HTTPException(status_code=404, detail="Task not found in the specified project")
|
||||
return db_task
|
||||
|
||||
router = APIRouter(prefix="/projects", tags=["projects"])
|
||||
|
||||
@router.get("/", response_model=List[ProjectBase], tags=["projects", "me"])
|
||||
##
|
||||
## GET endpoints
|
||||
##
|
||||
|
||||
@router.get("/", response_model=List[projects_schemas.ProjectBase], tags=["projects", "me"])
|
||||
def get_projects(db: db_dependency, request: Request):
|
||||
"""Get a user's projects"""
|
||||
|
||||
|
|
@ -42,65 +28,57 @@ def get_projects(db: db_dependency, request: Request):
|
|||
user_id = getattr(user, "id")
|
||||
|
||||
## fetching projects for the user
|
||||
projects = db.query(Project).join(Project.users).filter(User.id == user_id).all()
|
||||
projects = db.query(Project).join(Project.users).filter(getattr(users_schemas.UserBase, "id") == int(user_id)).all()
|
||||
return projects
|
||||
|
||||
|
||||
@router.get("/{project_id}", response_model=ProjectBase)
|
||||
@router.get("/{project_id}", response_model=projects_schemas.ProjectBase)
|
||||
def get_project(project_id: int, request:Request, db: db_dependency):
|
||||
"""Get a project by ID"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
project = get_project_by_id_for_user(user, project_id, db)
|
||||
return project
|
||||
|
||||
@router.get("/{project_id}/users", response_model=List[UserBase], tags=["users", "projects"])
|
||||
db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first()
|
||||
if db_project is None:
|
||||
raise HTTPException(status_code=404, detail="Project not found")
|
||||
if user not in db_project.users:
|
||||
raise HTTPException(status_code=403, detail="Not authorized to access this project")
|
||||
|
||||
return db_project
|
||||
|
||||
@router.get("/{project_id}/users", response_model=List[users_schemas.UserBase], tags=["users", "projects"])
|
||||
def get_project_users(project_id: int, request:Request, db: db_dependency):
|
||||
"""Get users from a specified project"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first()
|
||||
if db_project is None:
|
||||
raise HTTPException(status_code=404, detail="Project not found")
|
||||
if user not in db_project.users:
|
||||
raise HTTPException(status_code=403, detail="Not authorized to access this project's users")
|
||||
|
||||
return db_project.users
|
||||
|
||||
|
||||
@router.get("/{project_id}/tasks/{task_id}", response_model=TaskBase, tags=["tasks"])
|
||||
def get_project_task(project_id: int, task_id: int, db: db_dependency, request: Request):
|
||||
"""Get a specific task from a specified project"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
db_task = get_task_by_id_for_project(db_project, task_id, db)
|
||||
|
||||
return db_task
|
||||
|
||||
@router.get("/{project_id}/users/{user_id}", response_model=UserBase, tags=["users"])
|
||||
def get_project_user(project_id: int, user_id: int, db: db_dependency, request: Request):
|
||||
"""Get a specific user from a specified project"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
|
||||
db_project : ProjectBase = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
db_user = db.query(UserBase).filter(getattr(UserBase, "id") == user_id).first()
|
||||
if db_user is None or db_user not in db_project.users:
|
||||
raise HTTPException(status_code=404, detail="User not found in the specified project")
|
||||
return db_user
|
||||
|
||||
@router.get("/{project_id}/tasks", response_model=List[TaskBase], tags=["tasks", "projects"])
|
||||
@router.get("/{project_id}/tasks", response_model=List[tasks_schemas.TaskBase], tags=["tasks", "projects"])
|
||||
def get_project_tasks(project_id: int, request:Request, db: db_dependency):
|
||||
"""Get tasks from a specified project"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
db_tasks = db.query(TaskBase).filter(getattr(TaskBase, "project_id") == project_id).all()
|
||||
return db_tasks
|
||||
|
||||
@router.post("/", response_model=ProjectCreate)
|
||||
def create_project(project: ProjectCreate, request:Request, db: db_dependency):
|
||||
db_project = db.query(projects_schemas.ProjectBase).filter(getattr(projects_schemas.ProjectBase, "id") == project_id).first()
|
||||
if db_project is None:
|
||||
raise HTTPException(status_code=404, detail="Project not found")
|
||||
if user not in db_project.users:
|
||||
raise HTTPException(status_code=403, detail="Not authorized to access this project's tasks")
|
||||
|
||||
return db.query(tasks_schemas.TaskBase).filter(getattr(tasks_schemas.TaskBase, "project_id") == project_id).all()
|
||||
|
||||
@router.post("/", response_model=projects_schemas.ProjectCreate)
|
||||
def create_project(project: projects_schemas.ProjectCreate, request:Request, db: db_dependency):
|
||||
"""Create a new project"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
|
||||
db_project = ProjectCreate(
|
||||
db_project = projects_schemas.ProjectCreate(
|
||||
name=project.name,
|
||||
description=project.description,
|
||||
tasks=[],
|
||||
|
|
@ -115,121 +93,159 @@ def create_project(project: ProjectCreate, request:Request, db: db_dependency):
|
|||
|
||||
return db_project
|
||||
|
||||
# """Get tasks from a specified project"""
|
||||
# @router.get("/{project_id}/tasks", response_model=List[tasks_schemas.TaskBase], tags=["tasks"])
|
||||
# def read_tasks_from_project(project_id: int, db: db_dependency):
|
||||
# db_tasks = db.query(tasks.models.Task).filter(tasks.models.Task.project_id == project_id).all()
|
||||
# return db_tasks
|
||||
|
||||
@router.post("/{project_id}/tasks", response_model=ProjectTaskCreate, tags=["tasks"])
|
||||
def create_project_task(project_id: int, task: TaskCreate, db: db_dependency, request: Request):
|
||||
"""Create a new task in a specified project"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
# """Get a specific task from a specified project"""
|
||||
# @router.get("/{project_id}/tasks/{task_id}", response_model=tasks_schemas.TaskBase, tags=["tasks"])
|
||||
# def read_task_from_project(project_id: int, task_id: int, db: db_dependency):
|
||||
# db_task = db.query(tasks.models.Task).filter(tasks.models.Task.project_id == project_id, tasks.models.Task.id == task_id).first()
|
||||
# if db_task is None:
|
||||
# raise HTTPException(status_code=404, detail="Task not found in the specified project")
|
||||
# return db_task
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
db_task = ProjectTaskCreate(
|
||||
title=task.title,
|
||||
description=task.description,
|
||||
status=task.status,
|
||||
project=db_project
|
||||
)
|
||||
|
||||
db.add(db_task)
|
||||
db.commit()
|
||||
db.refresh(db_task)
|
||||
return db_task
|
||||
# """Get users from a specified project"""
|
||||
|
||||
@router.post("/{project_id}/users", response_model=ProjectAddUsers, tags=["users"])
|
||||
def add_project_user(project_id: int, user_data: ProjectAddUsers, db: db_dependency, request: Request):
|
||||
"""Add users to a specified project using their IDs"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
# @router.get("/{project_id}/users", response_model=List[users_schemas.UserBase], tags=["users"])
|
||||
# def read_users_from_project(project_id: int, db: db_dependency):
|
||||
# db_project = db.query(projects.models.Project).filter(projects.models.Project.id == project_id).first()
|
||||
# if db_project is None:
|
||||
# raise HTTPException(status_code=404, detail="Project not found")
|
||||
# return db_project.users
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
for user_id in user_data.user_ids:
|
||||
db_user = db.query(UserBase).filter(getattr(UserBase, "id") == user_id).first()
|
||||
if db_user:
|
||||
db_project.users.append(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_project)
|
||||
return db_project
|
||||
# ##
|
||||
# ## POST endpoints
|
||||
# ##
|
||||
|
||||
@router.delete("/{project_id}/users/{user_id}", response_model=ProjectRemoveUsers, tags=["users"])
|
||||
def remove_user_from_project(project_id: int, user_id: int, db: db_dependency, request: Request):
|
||||
"""Remove a user from a specified project using their ID"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
# """Create a new project"""
|
||||
|
||||
db_user = db.query(UserBase).filter(getattr(UserBase, "id") == user_id).first()
|
||||
if db_user is None or db_user not in db_project.users:
|
||||
raise HTTPException(status_code=404, detail="User not found in the specified project")
|
||||
# @router.post("/", response_model=projects.ProjectCreate)
|
||||
# def create_project(project: projects.ProjectCreate, db: db_dependency):
|
||||
# db_project = projects(
|
||||
# name=project.name,
|
||||
# description=project.description
|
||||
# )
|
||||
# db.add(db_project)
|
||||
# db.commit()
|
||||
# db.refresh(db_project)
|
||||
# return db_project
|
||||
|
||||
db_project.users.remove(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_project)
|
||||
return db_project
|
||||
|
||||
@router.put("/{project_id}/tasks/{task_id}", response_model=TaskUpdate, tags=["tasks"])
|
||||
def update_project_task(project_id: int, task_id: int, task: TaskUpdate, db: db_dependency, request: Request):
|
||||
"""Update a task in a specified project"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
db_task = get_task_by_id_for_project(db_project, task_id, db)
|
||||
# """Create a new task in a specified project"""
|
||||
|
||||
if task.title is not None:
|
||||
db_task.title = task.title
|
||||
if task.description is not None:
|
||||
db_task.description = task.description
|
||||
if task.status is not None:
|
||||
db_task.status = task.status
|
||||
# @router.post("/{project_id}/tasks", response_model=tasks.TaskBase, tags=["tasks"])
|
||||
# def create_task_in_project(project_id: int, task: tasks.TaskBase, db: db_dependency):
|
||||
# db_task = tasks.models.Task(
|
||||
# title=task.title,
|
||||
# description=task.description,
|
||||
# status=task.status,
|
||||
# project_id=project_id
|
||||
# )
|
||||
# db.add(db_task)
|
||||
# db.commit()
|
||||
# db.refresh(db_task)
|
||||
# return db_task
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_task)
|
||||
return db_task
|
||||
|
||||
@router.put("/{project_id}", response_model=ProjectUpdate)
|
||||
def update_project(project_id: int, project: ProjectUpdate, db: db_dependency, request: Request):
|
||||
"""Update a project by ID"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
# """Add users to a specified project using their IDs"""
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
# @router.post("/{project_id}/users", response_model=projects.ProjectAddUsers, tags=["users"])
|
||||
# def add_users_to_project(project_id: int, user_data: projects.ProjectAddUsers, db: db_dependency):
|
||||
# db_project = db.query(projects.models.Project).filter(projects.models.Project.id == project_id).first()
|
||||
# if db_project is None:
|
||||
# raise HTTPException(status_code=404, detail="Project not found")
|
||||
# for user_id in user_data.user_ids:
|
||||
# db_user = db.query(users.models.User).filter(users.models.User.id == user_id).first()
|
||||
# if db_user:
|
||||
# db_project.users.append(db_user)
|
||||
# db.commit()
|
||||
# db.refresh(db_project)
|
||||
# return db_project
|
||||
|
||||
if project.name is not None:
|
||||
db_project.name = project.name
|
||||
if project.description is not None:
|
||||
db_project.description = project.description
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_project)
|
||||
return db_project
|
||||
# ##
|
||||
# ## PUT endpoints
|
||||
# ##
|
||||
|
||||
@router.delete("/{project_id}", tags=["projects"])
|
||||
def delete_project(project_id: int, db: db_dependency, request: Request):
|
||||
"""Delete a project by ID"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
## Remove dangling tasks and user associations
|
||||
tasks: List[TaskBase] = db.query(TaskBase).filter(getattr(TaskBase, "project_id") == project_id).all()
|
||||
for task in tasks:
|
||||
db.delete(task)
|
||||
db_project.tasks.remove(task)
|
||||
|
||||
users: List[ProjectUserBase] = db.query(ProjectUserBase).join(ProjectUserBase).filter(getattr(ProjectBase, "id") == project_id).all()
|
||||
for proj_user in users:
|
||||
db_project.users.remove(proj_user)
|
||||
proj_user.projects.remove(db_project)
|
||||
# """Update a project by ID"""
|
||||
|
||||
db.delete(db_project)
|
||||
db.commit()
|
||||
return {"detail": "Project deleted successfully"}
|
||||
# @router.put("/{project_id}", response_model=projects.ProjectUpdate)
|
||||
# def update_project(project_id: int, project: projects.ProjectUpdate, db: db_dependency):
|
||||
# db_project = db.query(projects.models.Project).filter(projects.models.Project.id == project_id).first()
|
||||
# if db_project is None:
|
||||
# raise HTTPException(status_code=404, detail="Project not found")
|
||||
# if project.name is not None:
|
||||
# db_project.name = project.name
|
||||
# if project.description is not None:
|
||||
# db_project.description = project.description
|
||||
# db.commit()
|
||||
# db.refresh(db_project)
|
||||
# return db_project
|
||||
|
||||
@router.delete("/{project_id}/tasks/{task_id}" , tags=["tasks"])
|
||||
def delete_project_task(project_id: int, task_id: int, db: db_dependency, request: Request):
|
||||
"""Delete a task from a specified project"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
db_task = get_task_by_id_for_project(db_project, task_id, db)
|
||||
|
||||
db.delete(db_task)
|
||||
db_project.tasks.remove(db_task)
|
||||
|
||||
db.commit()
|
||||
return {"detail": "Task deleted successfully"}
|
||||
# """Update a task in a specified project"""
|
||||
|
||||
# @router.put("/{project_id}/tasks/{task_id}", response_model=tasks.TaskUpdate, tags=["tasks"])
|
||||
# def update_task_in_project(project_id: int, task_id: int, task: tasks.TaskUpdate, db: db_dependency):
|
||||
# db_task = db.query(tasks.models.Task).filter(tasks.models.Task.project_id == project_id, tasks.models.Task.id == task_id).first()
|
||||
# if db_task is None:
|
||||
# raise HTTPException(status_code=404, detail="Task not found in the specified project")
|
||||
# if task.title is not None:
|
||||
# db_task.title = task.title
|
||||
# if task.description is not None:
|
||||
# db_task.description = task.description
|
||||
# if task.status is not None:
|
||||
# db_task.status = task.status
|
||||
# db.commit()
|
||||
# db.refresh(db_task)
|
||||
# return db_task
|
||||
|
||||
# ##
|
||||
# ## DELETE endpoints
|
||||
# ##
|
||||
|
||||
# """Delete a project by ID"""
|
||||
|
||||
# @router.delete("/{project_id}")
|
||||
# def delete_project(project_id: int, db: db_dependency):
|
||||
# db_project = db.query(projects.models.Project).filter(projects.models.Project.id == project_id).first()
|
||||
# if db_project is None:
|
||||
# raise HTTPException(status_code=404, detail="Project not found")
|
||||
# db.delete(db_project)
|
||||
# db.commit()
|
||||
# return {"detail": "Project deleted successfully"}
|
||||
|
||||
|
||||
# """Delete a task from a specified project"""
|
||||
|
||||
# @router.delete("/{project_id}/tasks/{task_id}" , tags=["tasks"])
|
||||
# def delete_task_from_project(project_id: int, task_id: int, db: db_dependency):
|
||||
# db_task = db.query(tasks.models.Task).filter(tasks.models.Task.project_id == project_id, tasks.models.Task.id == task_id).first()
|
||||
# if db_task is None:
|
||||
# raise HTTPException(status_code=404, detail="Task not found in the specified project")
|
||||
# db.delete(db_task)
|
||||
# db.commit()
|
||||
# return {"detail": "Task deleted successfully"}
|
||||
|
||||
|
||||
# """Remove users from a specified project using their IDs"""
|
||||
|
||||
# @router.delete("/{project_id}/users/{user_id}", tags=["users"])
|
||||
# def remove_user_from_project(project_id: int, user_id: int, db: db_dependency):
|
||||
# db_project = db.query(projects.models.Project).filter(projects.models.Project.id == project_id).first()
|
||||
# if db_project is None:
|
||||
# raise HTTPException(status_code=404, detail="Project not found")
|
||||
# db_user = db.query(users.models.User).filter(users.models.User.id == user_id).first()
|
||||
# if db_user is None or db_user not in db_project.users:
|
||||
# raise HTTPException(status_code=404, detail="User not found in the specified project")
|
||||
# db_project.users.remove(db_user)
|
||||
# db.commit()
|
||||
# db.refresh(db_project)
|
||||
# return {"detail": "User removed from project successfully"}
|
||||
|
|
|
|||
|
|
@ -7,14 +7,12 @@ from schemas.users import UserBase
|
|||
class ProjectBase(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
name: str
|
||||
description: str
|
||||
tasks: List[TaskBase]
|
||||
users: List[UserBase]
|
||||
|
||||
class ProjectFull(ProjectBase):
|
||||
id: int
|
||||
|
||||
class ProjectCreate(BaseModel):
|
||||
name: str
|
||||
description: Optional[str] = None
|
||||
|
|
|
|||
|
|
@ -1,11 +0,0 @@
|
|||
import schemas.projects as project_schemas
|
||||
import schemas.tasks as task_schemas
|
||||
from pydantic import ConfigDict
|
||||
|
||||
class ProjectTaskBase(task_schemas.TaskBase):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
project: project_schemas.ProjectBase
|
||||
|
||||
class ProjectTaskCreate(task_schemas.TaskCreate):
|
||||
project: project_schemas.ProjectBase
|
||||
|
|
@ -17,11 +17,6 @@ class TaskBase(BaseModel):
|
|||
description: Optional[str] = None
|
||||
status: TaskStatus = TaskStatus.PENDING
|
||||
|
||||
class TaskCreate(BaseModel):
|
||||
title: str
|
||||
description: Optional[str] = None
|
||||
status: TaskStatus = TaskStatus.PENDING
|
||||
|
||||
class TaskUpdate(BaseModel):
|
||||
title: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue