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. "5275838fc73a1fc993fa9422a7fc1bc8d6a57b92" and "4722c4cf99964c7702043627865140ecf5fca25f" have entirely different histories.
5275838fc7
...
4722c4cf99
14 changed files with 0 additions and 914 deletions
8
.gitignore
vendored
8
.gitignore
vendored
|
|
@ -1,11 +1,3 @@
|
|||
*.db
|
||||
log.app
|
||||
|
||||
.idea/
|
||||
.vscode/
|
||||
|
||||
.env*
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[codz]
|
||||
|
|
|
|||
32
database.py
32
database.py
|
|
@ -1,32 +0,0 @@
|
|||
|
||||
import sqlalchemy
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.ext.declarative import declarative_base
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
|
||||
from fastapi import Depends
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import Annotated
|
||||
|
||||
URL_DATABASE = "sqlite:///./kanban_clone.db"
|
||||
|
||||
engine = create_engine(URL_DATABASE, echo=True)
|
||||
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
Base = declarative_base()
|
||||
|
||||
def init_db() -> None:
|
||||
# Import models so they are registered with SQLAlchemy metadata
|
||||
import models # noqa: F401
|
||||
Base.metadata.create_all(bind=engine)
|
||||
|
||||
def get_db():
|
||||
db = SessionLocal()
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
db_dependency = Annotated[Session, Depends(get_db)]
|
||||
153
main.py
153
main.py
|
|
@ -1,153 +0,0 @@
|
|||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI, HTTPException
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from routers.projects import router as projects_router
|
||||
|
||||
from routers.users import router as users_router
|
||||
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).
|
||||
|
||||
## Other projects
|
||||
Here are some frontend implementations for this API:
|
||||
- [KanbanCloneAngular](https://github.com/a-mayb3/KanbanCloneAngular) - Angular frontend
|
||||
- [KanbanCloneAndroid](https://github.com/a-mayb3/KanbanCloneAndroid) - Android frontend
|
||||
|
||||
|
||||
"""
|
||||
|
||||
global_logger = logging.getLogger()
|
||||
global_logger.setLevel(logging.INFO)
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
|
||||
filename='app.log',
|
||||
encoding='utf-8'
|
||||
)
|
||||
handler = logging.StreamHandler()
|
||||
formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
|
||||
handler.setFormatter(formatter)
|
||||
global_logger.addHandler(handler)
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
logger = global_logger
|
||||
|
||||
# Place for startup and shutdown events if needed in the future
|
||||
logger.info("Initializing database...")
|
||||
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.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["http://localhost:3000", "http://localhost:5173"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
|
||||
app.include_router(auth_router)
|
||||
app.include_router(users_router)
|
||||
app.include_router(me_router)
|
||||
app.include_router(projects_router)
|
||||
|
||||
"""ping pong :)"""
|
||||
@app.get("/ping")
|
||||
def ping():
|
||||
return {"message": "pong"}
|
||||
|
||||
"""Gives project url"""
|
||||
@app.get("/sources")
|
||||
def source():
|
||||
return {"url": "https://github.com/a-mayb3/Kanban_clone_backend"}
|
||||
|
||||
if __name__ == "__main__":
|
||||
import uvicorn
|
||||
uvicorn.run(app, host="0.0.0.0", port=8000)
|
||||
|
||||
|
||||
from fastapi.exceptions import RequestValidationError
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
@app.exception_handler(HTTPException)
|
||||
async def http_exception_handler(request, exc):
|
||||
"""Custom HTTP exception handler"""
|
||||
|
||||
logger = global_logger
|
||||
logger.error(f"HTTP error occurred: {exc.detail}")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=exc.status_code,
|
||||
content={
|
||||
"error": {
|
||||
"message": exc.detail,
|
||||
"type": "authentication_error" if exc.status_code == 401 else "authorization_error",
|
||||
"status_code": exc.status_code
|
||||
}
|
||||
},
|
||||
headers=exc.headers
|
||||
)
|
||||
|
||||
@app.exception_handler(RequestValidationError)
|
||||
async def validation_exception_handler(request, exc):
|
||||
"""Handle validation errors"""
|
||||
|
||||
logger = global_logger
|
||||
logger.error(f"Validation error: {exc.errors()}")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=422,
|
||||
content={
|
||||
"error": {
|
||||
"message": "Validation error",
|
||||
"type": "validation_error",
|
||||
"details": exc.errors()
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def general_exception_handler(request, exc):
|
||||
"""Handle all other exceptions"""
|
||||
|
||||
logger = global_logger
|
||||
logger.error(f"Unexpected error: {exc}")
|
||||
|
||||
return JSONResponse(
|
||||
status_code=500,
|
||||
content={
|
||||
"error": {
|
||||
"message": "An unexpected error occurred.",
|
||||
"type": "internal_server_error",
|
||||
"details": str(exc)
|
||||
}
|
||||
}
|
||||
)
|
||||
41
models.py
41
models.py
|
|
@ -1,41 +0,0 @@
|
|||
from sqlalchemy import Column, ForeignKey, String, Integer, Table
|
||||
from sqlalchemy.dialects.sqlite import BLOB
|
||||
from sqlalchemy.orm import relationship
|
||||
from database import Base
|
||||
from typing import Optional, List
|
||||
|
||||
project_user = Table(
|
||||
"project_user",
|
||||
Base.metadata,
|
||||
Column("project_id", Integer, ForeignKey("projects.id"), primary_key=True),
|
||||
Column("user_id", Integer, ForeignKey("users.id"), primary_key=True)
|
||||
)
|
||||
|
||||
class User(Base):
|
||||
__tablename__ = "users"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
|
||||
name = Column(String, index=True)
|
||||
email = Column(String, unique=True, index=True)
|
||||
password_hash = Column(String)
|
||||
password_salt = Column(String)
|
||||
projects = relationship("Project", secondary=project_user, back_populates="users")
|
||||
|
||||
class Project(Base):
|
||||
__tablename__ = "projects"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
|
||||
name = Column(String, index=True)
|
||||
description = Column(String)
|
||||
users = relationship("User", secondary=project_user, back_populates="projects")
|
||||
tasks = relationship("Task", back_populates="project")
|
||||
|
||||
class Task(Base):
|
||||
__tablename__ = "tasks"
|
||||
|
||||
id = Column(Integer, primary_key=True, index=True, autoincrement=True)
|
||||
title = Column(String, index=True)
|
||||
description = Column(String)
|
||||
status = Column(String, default="pending")
|
||||
project_id = Column(Integer, ForeignKey("projects.id"))
|
||||
project = relationship("Project", back_populates="tasks")
|
||||
|
|
@ -1,58 +0,0 @@
|
|||
annotated-doc==0.0.4
|
||||
annotated-types==0.7.0
|
||||
anyio==4.12.1
|
||||
bcrypt==5.0.0
|
||||
certifi==2026.1.4
|
||||
cffi==2.0.0
|
||||
click==8.3.1
|
||||
cryptography==46.0.4
|
||||
dnspython==2.8.0
|
||||
ecdsa==0.19.1
|
||||
email-validator==2.3.0
|
||||
fastapi==0.128.0
|
||||
fastapi-cli==0.0.20
|
||||
fastapi-cloud-cli==0.11.0
|
||||
fastar==0.8.0
|
||||
greenlet==3.3.1
|
||||
h11==0.16.0
|
||||
httpcore==1.0.9
|
||||
httptools==0.7.1
|
||||
httpx==0.28.1
|
||||
idna==3.11
|
||||
itsdangerous==2.2.0
|
||||
Jinja2==3.1.6
|
||||
markdown-it-py==4.0.0
|
||||
MarkupSafe==3.0.3
|
||||
mdurl==0.1.2
|
||||
passlib==1.7.4
|
||||
pyargon2==1.1.2
|
||||
pyasn1==0.6.2
|
||||
pycparser==3.0
|
||||
pydantic==2.12.5
|
||||
pydantic-extra-types==2.11.0
|
||||
pydantic-settings==2.12.0
|
||||
pydantic_core==2.41.5
|
||||
Pygments==2.19.2
|
||||
python-dotenv==1.2.1
|
||||
python-jose==3.5.0
|
||||
python-multipart==0.0.22
|
||||
PyYAML==6.0.3
|
||||
rich==14.3.1
|
||||
rich-toolkit==0.17.1
|
||||
rignore==0.7.6
|
||||
rsa==4.9.1
|
||||
sentry-sdk==2.50.0
|
||||
shellingham==1.5.4
|
||||
six==1.17.0
|
||||
SQLAlchemy==2.0.46
|
||||
starlette==0.50.0
|
||||
starlette-session==0.4.3
|
||||
starlette-session-middleware==0.1.6
|
||||
typer==0.21.1
|
||||
typing-inspection==0.4.2
|
||||
typing_extensions==4.15.0
|
||||
urllib3==2.6.3
|
||||
uvicorn==0.40.0
|
||||
uvloop==0.22.1
|
||||
watchfiles==1.1.1
|
||||
websockets==16.0
|
||||
142
routers/auth.py
142
routers/auth.py
|
|
@ -1,142 +0,0 @@
|
|||
import os
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, status, Response
|
||||
from database import db_dependency
|
||||
from jose import JWTError, jwt
|
||||
from datetime import datetime, timedelta, timezone
|
||||
import models
|
||||
|
||||
import schemas.users as user_schemas
|
||||
|
||||
from pyargon2 import hash
|
||||
|
||||
router = APIRouter(prefix="/auth", tags=["auth"])
|
||||
|
||||
SECRET_KEY = os.getenv("SECRET_KEY", "your-secret-key-change-this-in-production")
|
||||
ALGORITHM = "HS256"
|
||||
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
|
||||
else:
|
||||
expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
|
||||
to_encode.update({"exp": expire})
|
||||
to_encode.update({"iat": datetime.now(timezone.utc)})
|
||||
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:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Not logged in"
|
||||
)
|
||||
try:
|
||||
user_id: str = verify_jwt_token(get_token) ## verifying token validity
|
||||
|
||||
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"
|
||||
)
|
||||
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:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail="Could not verify credentials")
|
||||
|
||||
@router.post("/login")
|
||||
def login(user_data: user_schemas.UserLogin, request: Request, response: Response, db: db_dependency):
|
||||
"""Login and receive JWT token in cookie"""
|
||||
|
||||
## check if access token already exists
|
||||
get_token = request.cookies.get("access_token")
|
||||
if get_token:
|
||||
try:
|
||||
user_id = verify_jwt_token(get_token)
|
||||
return {
|
||||
"message": "Already logged in",
|
||||
"user": {
|
||||
"id": user_id
|
||||
}
|
||||
}
|
||||
except HTTPException:
|
||||
request.cookies.clear() ## removing invalid auth cookie
|
||||
|
||||
## 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(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect email or password"
|
||||
)
|
||||
|
||||
verify_user_password(getattr(db_user, "id"), user_data.password, db)
|
||||
|
||||
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
|
||||
)
|
||||
|
||||
# Set JWT in httpOnly cookie
|
||||
response.set_cookie(
|
||||
key="access_token",
|
||||
value=access_token,
|
||||
httponly=True,
|
||||
max_age=ACCESS_TOKEN_EXPIRE_MINUTES * 60,
|
||||
samesite="lax",
|
||||
secure=False # Set to True in production with HTTPS
|
||||
)
|
||||
|
||||
return {
|
||||
"message": "Login successful",
|
||||
"user": {
|
||||
"id": db_user.id,
|
||||
"name": db_user.name,
|
||||
"email": db_user.email
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1,57 +0,0 @@
|
|||
from fastapi import APIRouter, Depends, HTTPException, status, Response, Request
|
||||
from database import db_dependency
|
||||
from jose import JWTError, jwt
|
||||
import models
|
||||
|
||||
from routers import auth
|
||||
|
||||
from schemas.users import UserBase
|
||||
from schemas.projects import ProjectBase
|
||||
from schemas.projects_users import ProjectUserBase
|
||||
|
||||
|
||||
router = APIRouter(prefix="/me", tags=["me"])
|
||||
|
||||
@router.get("/", response_model=ProjectUserBase, tags=["me", "users"])
|
||||
def get_me(request: Request, db: db_dependency):
|
||||
"""Get current authenticated user"""
|
||||
user = auth.get_user_from_jwt(request, db)
|
||||
return user
|
||||
|
||||
|
||||
@router.get("/logout", tags=["me", "auth"])
|
||||
def logout(request: Request,response: Response):
|
||||
"""Logout by clearing the JWT cookie"""
|
||||
|
||||
get_token = request.cookies.get("access_token")
|
||||
if not get_token:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Not logged in"
|
||||
)
|
||||
|
||||
response.delete_cookie(key="access_token")
|
||||
return {"message": "Logout successful"}
|
||||
|
||||
@router.delete("/delete-me", tags=["me", "auth", "users"])
|
||||
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
|
||||
request.cookies.clear()
|
||||
return {"message": "User deleted successfully"}
|
||||
|
|
@ -1,234 +0,0 @@
|
|||
from fastapi import APIRouter, HTTPException, Depends, Request
|
||||
from typing import List, Annotated
|
||||
|
||||
from database import db_dependency
|
||||
|
||||
from schemas.tasks import TaskBase, TaskCreate, TaskUpdate
|
||||
from schemas.projects import ProjectBase, ProjectCreate, ProjectUpdate, ProjectAddUser, ProjectRemoveUsers, ProjectFull
|
||||
from schemas.users import UserBase
|
||||
from schemas.projects_users import ProjectUserBase
|
||||
from schemas.projects_tasks import ProjectTaskBase, ProjectTaskCreate
|
||||
|
||||
from models import Project, Task, User
|
||||
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[ProjectFull], tags=["projects", "me"])
|
||||
def get_projects(db: db_dependency, request: Request):
|
||||
"""Get a user's projects"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
user_id = getattr(user, "id")
|
||||
|
||||
## fetching projects for the user
|
||||
projects = db.query(Project).join(Project.users).filter(User.id == user_id).all()
|
||||
return projects
|
||||
|
||||
|
||||
@router.get("/{project_id}", response_model=ProjectFull)
|
||||
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"])
|
||||
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)
|
||||
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(User).filter(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")
|
||||
return db_user
|
||||
|
||||
@router.get("/{project_id}/tasks", response_model=List[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(Task).filter(Task.project_id == project_id).all()
|
||||
return db_tasks
|
||||
|
||||
@router.post("/", response_model=ProjectCreate)
|
||||
def create_project(project: ProjectCreate, request:Request, db: db_dependency):
|
||||
"""Create a new project"""
|
||||
|
||||
user = get_user_from_jwt(request, db)
|
||||
|
||||
db_project = Project(
|
||||
name=project.name,
|
||||
description=project.description,
|
||||
tasks=[]
|
||||
)
|
||||
|
||||
db_project.users.append(user)
|
||||
|
||||
db.add(db_project)
|
||||
db.commit()
|
||||
db.refresh(db_project)
|
||||
|
||||
return db_project
|
||||
|
||||
|
||||
@router.post("/{project_id}/tasks", response_model=ProjectTaskBase, 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)
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
db_task = Task(
|
||||
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
|
||||
|
||||
@router.post("/{project_id}/users", response_model=ProjectFull, tags=["users"])
|
||||
def add_project_user(project_id: int, user_data: ProjectAddUser, db: db_dependency, request: Request):
|
||||
"""Add a user to a specified project using their email address"""
|
||||
user = get_user_from_jwt(request, db)
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
db_user = db.query(User).filter(User.email == user_data.user_email).first()
|
||||
|
||||
if not db_user:
|
||||
raise HTTPException(status_code=404, detail="User with the specified email not found")
|
||||
|
||||
if db_user not in db_project.users:
|
||||
db_project.users.append(db_user)
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="User is already a member of the project")
|
||||
|
||||
db.commit()
|
||||
db.refresh(db_project)
|
||||
return db_project
|
||||
|
||||
@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)
|
||||
|
||||
db_user = db.query(User).filter(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 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)
|
||||
|
||||
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
|
||||
|
||||
@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)
|
||||
|
||||
db_project = get_project_by_id_for_user(user, project_id, db)
|
||||
|
||||
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}", 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[Task] = db.query(Task).filter(Task.project_id == project_id).all()
|
||||
for task in tasks:
|
||||
db.delete(task)
|
||||
db_project.tasks.remove(task)
|
||||
|
||||
db.delete(db_project)
|
||||
db.commit()
|
||||
return {"detail": "Project deleted successfully"}
|
||||
|
||||
@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"}
|
||||
|
|
@ -1,76 +0,0 @@
|
|||
import os
|
||||
from typing import List
|
||||
from fastapi import APIRouter, HTTPException, Depends, Request
|
||||
from jose import JWTError, jwt
|
||||
from database import db_dependency
|
||||
|
||||
import models
|
||||
|
||||
from routers import auth
|
||||
import schemas.users as users
|
||||
import schemas.projects as projects
|
||||
from routers.auth import get_user_from_jwt
|
||||
|
||||
from pyargon2 import hash
|
||||
|
||||
router = APIRouter(prefix="/users", tags=["users"])
|
||||
|
||||
|
||||
@router.get("/{user_id}", response_model=users.UserBase)
|
||||
def read_user(user_id: int, db: db_dependency, request:Request):
|
||||
"""Get a user by ID"""
|
||||
|
||||
get_user_from_jwt(request, db)
|
||||
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user
|
||||
|
||||
@router.get("/{user_id}/projects", response_model=List[projects.ProjectBase])
|
||||
def read_projects_from_user(user_id: int, db: db_dependency, request: Request):
|
||||
"""Get projects assigned to a user"""
|
||||
|
||||
get_user_from_jwt(request, db)
|
||||
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
return db_user.projects
|
||||
|
||||
##
|
||||
## POST endpoints
|
||||
##
|
||||
|
||||
@router.post("/", response_model=users.UserBase)
|
||||
def create_user(user: users.UserCreate, db: db_dependency):
|
||||
"""Create a new user"""
|
||||
|
||||
user_salt = os.urandom(32).hex()
|
||||
print("Generated salt:", user_salt)
|
||||
|
||||
hashed_password = hash(password=user.password, salt=user_salt, variant="id")
|
||||
|
||||
db_user = models.User(
|
||||
name=user.name,
|
||||
email=user.email,
|
||||
password_hash=hashed_password,
|
||||
password_salt=user_salt
|
||||
)
|
||||
|
||||
db.add(db_user)
|
||||
db.commit()
|
||||
db.refresh(db_user)
|
||||
return db_user
|
||||
|
||||
@router.delete("/{user_id}")
|
||||
def delete_user(user_id: int, db: db_dependency):
|
||||
db_user = db.query(models.User).filter(models.User.id == user_id).first()
|
||||
if db_user is None:
|
||||
raise HTTPException(status_code=404, detail="User not found")
|
||||
db.delete(db_user)
|
||||
db.commit()
|
||||
return {"detail": "User deleted"}
|
||||
|
||||
|
||||
|
||||
|
|
@ -1,36 +0,0 @@
|
|||
from pydantic import BaseModel, ConfigDict
|
||||
from typing import List, Optional
|
||||
|
||||
from schemas.tasks import TaskBase
|
||||
from schemas.users import UserBase
|
||||
|
||||
class ProjectBase(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
name: str
|
||||
description: str
|
||||
tasks: List[TaskBase]
|
||||
users: List[UserBase]
|
||||
|
||||
class ProjectFull(ProjectBase):
|
||||
id: int
|
||||
|
||||
class ProjectCreate(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
name: str
|
||||
description: Optional[str] = None
|
||||
tasks: List[TaskBase] = []
|
||||
|
||||
class ProjectUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
description: Optional[str] = None
|
||||
|
||||
class ProjectAddUser(BaseModel):
|
||||
user_email: str
|
||||
|
||||
class ProjectAddUsers(BaseModel):
|
||||
user_ids: List[int] = []
|
||||
|
||||
class ProjectRemoveUsers(BaseModel):
|
||||
user_ids: List[int] = []
|
||||
|
|
@ -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
|
||||
|
|
@ -1,11 +0,0 @@
|
|||
from typing import List
|
||||
|
||||
from pydantic import ConfigDict
|
||||
from schemas.projects import ProjectFull
|
||||
from schemas.users import UserBase
|
||||
|
||||
class ProjectUserBase(UserBase):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
projects: List[ProjectFull]
|
||||
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
from enum import Enum
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from typing import List, Annotated, Optional
|
||||
|
||||
class TaskStatus(str, Enum):
|
||||
PENDING = "pending"
|
||||
IN_PROGRESS = "in_progress"
|
||||
COMPLETED = "completed"
|
||||
FAILED = "failed"
|
||||
STASHED = "stashed"
|
||||
|
||||
class TaskBase(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
title: str
|
||||
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
|
||||
status: Optional[TaskStatus] = None
|
||||
|
||||
|
|
@ -1,26 +0,0 @@
|
|||
from pydantic import BaseModel, ConfigDict
|
||||
from typing import List, Optional
|
||||
|
||||
class UserBase(BaseModel):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
|
||||
id: int
|
||||
name: str
|
||||
email: str
|
||||
|
||||
class UserCreate(BaseModel):
|
||||
name: str
|
||||
email: str
|
||||
password: str
|
||||
|
||||
class UserUpdateInfo(BaseModel):
|
||||
name: Optional[str] = None
|
||||
email: Optional[str] = None
|
||||
|
||||
class UserUpdatePassword(BaseModel):
|
||||
password: str
|
||||
new_password: str
|
||||
|
||||
class UserLogin(BaseModel):
|
||||
email: str
|
||||
password: str
|
||||
Loading…
Add table
Add a link
Reference in a new issue