Coverage for app / main.py: 90%

1import time 

2 

3from fastapi import FastAPI, Depends 

4from fastapi.middleware.cors import CORSMiddleware 

5from contextlib import asynccontextmanager 

6from datetime import datetime 

7from typing import List, Optional 

8from pydantic import BaseModel, ConfigDict 

9from sqlalchemy.orm import Session 

10from sqlalchemy import create_engine 

11from sqlalchemy.exc import OperationalError 

12 

13from app.database import Base, engine, get_db, DATABASE_URL 

14 

15# These are required so that when the database is created, the correct tables are added 

16from app.models.user import User 

17from app.models.task import Task 

18from app.models.subtask import Subtask 

19from app.schemas import TaskResponse 

20from app.models.moodleTask import MoodleTask 

21# from app.models.reminders import Reminders 

22# from app.models.notification import Notification 

23# from app.scheduler import start_scheduler, stop_scheduler 

24 

25from app.tasks import router as tasks_router 

26# from app.reminders import router as reminders_router 

27from app.moodleTasks import router as moodletasks_router 

28 

29engine = create_engine(DATABASE_URL) 

30while True: 

31 try: 

32 conn = engine.connect() 

33 conn.close() 

34 break 

35 except OperationalError: 

36 print("Database not ready, retrying in 2s...") 

37 time.sleep(2) 

38 

39Base.metadata.create_all(bind=engine) 

40 

41# @asynccontextmanager 

42# async def lifespan(app: FastAPI): 

43# # Startup 

44# start_scheduler() 

45# yield 

46# # Shutdown 

47# stop_scheduler() 

48 

49app = FastAPI() 

50 

51@app.get("/") 

52async def root(): 

53 return {"message": "Hello World"} 

54 

55 

56app.include_router(tasks_router, prefix="/tasks", tags=["tasks"]) 

57# app.include_router(reminders_router, prefix="/reminders", tags=["reminders"]) 

58app.include_router(moodletasks_router, prefix="/moodletasks", tags=["moodletasks"]) 

59class TaskResponse(BaseModel): 

60 id: int 

61 user_id: int 

62 title: str # The title of the task 

63 description: Optional[str] = None # A short description of the task 

64 completed: bool # Whether or not the task is complete 

65 importance: int # How important the task is (scale from 1-10) 

66 length: int # How many minuites this will take (<5 - 300) 

67 tags: List[str] = ( 

68 [] 

69 ) # A list of string tags (can be []). No longer than 50 chars per tag 

70 due_at: Optional[datetime] = None # The date that this must be completed by 

71 created_at: datetime 

72 updated_at: datetime 

73 

74 model_config = ConfigDict(from_attributes=True) 

75 

76 

77@app.get( 

78 "/tasks", 

79 response_model=List[TaskResponse], 

80 summary="Retrieve all tasks", 

81 description="Fetches a list of all tasks from the database, including their ID, title, description, and completion status.", 

82) 

83def get_tasks(db: Session = Depends(get_db)): 

84 tasks = db.query(Task).all() 

85 return tasks