ns-task-list-api

ada-project-docs/wave_04.md

@@ -193,4 +193,4 @@ Send `PATCH` requests to `localhost:5000/tasks/<task_id>/mark_complete` (use the
 
 ![](assets/postman_patch.png)
 
-![](assets/slack_notification_feature.png)
+![](assets/slIntegack_notification_feature.png)

app/__init__.py

@@ -1,5 +1,7 @@
 from flask import Flask
 from .db import db, migrate
+from .routes.task_routes import bp as tasks_bp
+from .routes.goal_routes import bp as goals_bp
 from .models import task, goal
 import os
 
@@ -10,13 +12,13 @@ def create_app(config=None):
     app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('SQLALCHEMY_DATABASE_URI')
 
     if config:
-        # Merge `config` into the app's configuration
-        # to override the app's default settings for testing
         app.config.update(config)
 
     db.init_app(app)
     migrate.init_app(app, db)
 
-    # Register Blueprints here
+    app.register_blueprint(tasks_bp)
+    app.register_blueprint(goals_bp)
 
     return app
+

app/models/goal.py

@@ -1,5 +1,17 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from ..db import db
+# from app.models.task import Task
 
 class Goal(db.Model):
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    title: Mapped[str] 
+    tasks: Mapped[list["Task"]] = relationship("Task", back_populates="goal")
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "title": self.title
+        }
+    @classmethod
+    def from_dict(cls, goal_data):
+        new_goal = cls(title=goal_data["title"])
+        return new_goal

app/models/task.py

@@ -1,5 +1,39 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from ..db import db
+from datetime import datetime
+from sqlalchemy import ForeignKey
+from typing import Optional
 
 class Task(db.Model):
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    title: Mapped[str] = mapped_column(nullable=False)
+    description: Mapped[Optional[str]] = mapped_column(nullable=True)
+    completed_at: Mapped[Optional[datetime]] = mapped_column(nullable=True)
+    goal_id: Mapped[Optional[int]] = mapped_column(ForeignKey("goal.id"))
+    goal: Mapped[Optional["Goal"]] = relationship("Goal", back_populates="tasks")
+
+
+    def to_dict(self, include_goal=False, include_goal_id=False):
+        task_dict = dict(
+            id=self.id,
+            title=self.title,
+            description=self.description,
+            is_complete=self.completed_at is not None 
+        )
+        if include_goal_id:
+            task_dict["goal_id"] = self.goal_id
+
+        if include_goal and self.goal:
+            task_dict['goal'] = self.goal.title 
+
+
+        return task_dict
+
+    @classmethod
+    def from_dict(cls, task_data):
+        return cls(
+            title = task_data["title"],
+            description = task_data["description"],
+            completed_at = task_data.get("completed_at"),
+            goal_id=task_data.get("goal_id", None)
+        )

app/routes/goal_routes.py

@@ -1 +1,160 @@
-from flask import Blueprint
+from flask import Blueprint, request, abort, make_response
+from app.models.goal import Goal
+from ..db import db
+from .route_utilities import validate_model, get_model_with_filters, create_model
+
+
+
+bp = Blueprint("goals_bp", __name__, url_prefix="/goals")
+
+@bp.post("")
+def create_goal():
+    request_body = request.get_json()
+    if "title" not in request_body:
+        abort(make_response({"details": "Invalid data"}, 400))
+
+    new_goal = Goal.from_dict(request_body)
+    db.session.add(new_goal)
+    db.session.commit()
+
+    response = {
+        "goal": new_goal.to_dict()  
+    }
+    return response, 201
+
+@bp.post("/<goal_id>/tasks")
+def create_task_with_goal_id(goal_id):
+    from app.models.goal import Goal
+    from app.models.task import Task
+
+    goal = validate_model(Goal, goal_id)
+    if not goal:
+        response = {"message": f"Goal with ID {goal_id} not found"}
+        abort(make_response(response, 404))
+
+    request_body = request.get_json()
+    task_ids = request_body.get("task_ids")
+
+    if not task_ids:
+        response = {"message": "Invalid request: missing 'task_ids'"}
+        abort(make_response(response, 400))
+
+    tasks = Task.query.filter(Task.id.in_(task_ids)).all()
+
+    if len(tasks) != len(task_ids):
+        response = {"message": "One or more tasks not found"}
+        abort(make_response(response, 404))
+
+    for task in tasks:
+        if task not in goal.tasks:
+            goal.tasks.append(task)
+
+    db.session.commit()
+
+    goal_data = {
+        "id": goal.id,
+        "task_ids": [task.id for task in goal.tasks]
+    }
+
+    response = goal_data  
+    return response, 200
+
+
+@bp.get("/<goal_id>/tasks")
+def get_tasks_for_goal(goal_id):
+    from app.models.goal import Goal
+    from app.models.task import Task
+    goal = validate_model(Goal, goal_id)
+
+
+    if goal.tasks:  
+        task_data = [task.to_dict() for task in goal.tasks]
+    else: 
+        tasks = Task.query.filter_by(goal_id=goal.id).all()
+        task_data = [
+            task.to_dict() for task in tasks
+        ]
+
+    task_data = task_data or []
+
+    for task in task_data:
+        task["is_complete"] = bool(task.get("is_complete", False))  # Explicitly cast to bool
+
+    response = {
+        "id": goal.id,
+        "title": goal.title,
+        "tasks": task_data 
+    }
+
+    return response, 200
+
+@bp.get("/tasks/<task_id>")
+def get_task_by_id(task_id):
+    from app.models.task import Task
+    from app.models.goal import Goal
+
+    # Validate that the task exists
+    task = Task.query.get(task_id)
+    if not task:
+        response = {"message": f"task id {task_id} not found"}
+        abort(make_response(response, 404))
+
+    task_data = {
+        "id": task.id,
+        "goal_id": task.goal_id,  
+        "title": task.title,
+        "description": task.description,
+        "is_complete": task.completed_at is not None 
+    }
+
+    return {"task": task_data}, 200
+
+
+
+@bp.get("")
+def get_all_goal():
+    return get_model_with_filters(Goal, request.args)
+
+@bp.get("/<goal_id>")
+def get_one_goal(goal_id):
+    goal = validate_model(Goal, goal_id)
+    if not goal:
+        abort(make_response({"message": f"goal_id {goal_id} not found"}, 404))
+    goal_response = {
+        "goal": goal.to_dict()  
+    }
+
+    return goal_response, 200
+
+
+@bp.put("/<goal_id>")
+def update_goal(goal_id):
+    goal = validate_model(Goal, goal_id)
+    if not goal:
+        abort(make_response({"message": f"goal_id {goal_id} not found"}, 404))
+    request_body = request.get_json()
+
+    goal.title = request_body["title"]
+
+    db.session.commit()
+
+    response = {
+        "goal": goal.to_dict()  
+    }
+
+    return response, 200
+
+@bp.delete("/<goal_id>")
+def delete_goal(goal_id):
+    goal = validate_model(Goal, goal_id)
+    if not goal:
+        abort(make_response({"message": f"goal_id {goal_id} not found"}, 404))
+
+    db.session.delete(goal)
+    db.session.commit()
+
+    response = {
+        "details": f'Goal {goal.id} "{goal.title}" successfully deleted'
+    }
+
+    return response, 200

app/routes/route_utilities.py

@@ -0,0 +1,40 @@
+from flask import abort, make_response
+from ..db import db
+
+def validate_model(cls, model_id):
+    model_prefix = 'goal' if cls.__name__.lower() == 'goal' else 'task'
+
+    try:
+        model_id = int(model_id)
+    except ValueError:
+        abort(make_response({"message": f"{model_prefix}_id {model_id} invalid"}, 400))
+
+    # query = db.select(cls).where(cls.id == model_id)
+    # model = db.session.scalar(query)
+    model = cls.query.get(model_id)
+    if not model:
+        abort(make_response({"message": f"{model_prefix}_id {model_id} not found"}, 404))
+
+    return model
+
+
+def create_model(cls, model_data):
+    try:
+        new_model = cls.from_dict(model_data)
+    except KeyError as error:
+        response = {"message": f"Invalid request: missing {error.args[0]}"}
+        abort(make_response(response, 400))
+
+def get_model_with_filters(cls, filters=None):
+    query = db.select(cls)
+
+    if filters:
+        for attribute, value in filters.items():
+            if isinstance(value, list):
+                    query = query.where(getattr(cls, attribute).in_(value))
+            else:
+                    query = query.where(getattr(cls, attribute).ilike(f"%{value}%") if isinstance(value, str) else getattr(cls, attribute) == value)
+    models = db.session.scalars(query.order_by(cls.id))
+    models_response = [model.to_dict() for model in models]
+
+    return models_response