Crow-BCaprirolo-Task-List-API

app/__init__.py

@@ -1,10 +1,15 @@
 from flask import Flask
+from flask_cors import CORS
 from .db import db, migrate
 from .models import task, goal
+from .routes.task_routes import bp as tasks_bp
+from .routes.goal_routes import bp as goals_bp
 import os
 
 def create_app(config=None):
     app = Flask(__name__)
+    CORS(app)
+    app.config['CORS_HEADERS'] = 'Content-Type'
 
     app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
     app.config['SQLALCHEMY_DATABASE_URI'] = os.environ.get('SQLALCHEMY_DATABASE_URI')
@@ -18,5 +23,7 @@ def create_app(config=None):
     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,25 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
 from ..db import db
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .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(back_populates="goal")
+
+    def to_dict(self):
+        goal_as_dict = {
+            "id": self.id,
+            "title": self.title
+            }
+
+        return goal_as_dict
+
+    @classmethod
+    def from_dict(cls, task_data):
+        new_task =  cls(title=task_data['title'])
+
+        return new_task

app/models/task.py

@@ -1,5 +1,50 @@
-from sqlalchemy.orm import Mapped, mapped_column
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+from sqlalchemy import ForeignKey
 from ..db import db
+from datetime import datetime
+from typing import Optional, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from .goal import Goal
+
 
 class Task(db.Model):
     id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
+    title: Mapped[str]
+    description: Mapped[str]
+    goal_id: Mapped[Optional[int]] = mapped_column(ForeignKey("goal.id"))
+    goal: Mapped[Optional["Goal"]] = relationship(back_populates="tasks")
+
+    completed_at: Mapped[Optional[datetime]]
+
+    #converts task instance into dictionary
+    def to_dict(self):
+        task_as_dict = {}
+        task_as_dict["id"]= self.id
+        task_as_dict["title"]= self.title
+        task_as_dict["description"]= self.description
+        task_as_dict["is_complete"]= self.completed_at is not None
+
+        if self.goal_id:
+            task_as_dict["goal_id"] = self.goal_id
+
+        if self.completed_at:
+            task_as_dict["completed_at"]= self.completed_at.isoformat()
+        else:
+            None
+
+        return task_as_dict
+
+    #creates a task from a dictionary
+    @classmethod
+    def from_dict(cls, task_data):
+
+        new_task =  cls(
+            title=task_data['title'],
+            description=task_data['description'],
+            completed_at=task_data.get('completed_at'),
+            goal_id=task_data.get('goal_id')
+            )
+
+        return new_task
+

app/routes/goal_routes.py

@@ -1 +1,89 @@
-from flask import Blueprint
+from flask import Blueprint, request
+from app.models.goal import Goal
+from app.models.task import Task
+
+from .route_utilities import *
+
+bp = Blueprint("goals_bp", __name__, url_prefix="/goals")
+
+def associate_tasks_with_goal(goal, task_ids):
+
+    for task in goal.tasks:
+        task.goal_id = None
+
+    #add new tasks
+    for task_id in task_ids:
+        task = validate_model(Task, task_id)
+        task.goal_id = goal.id
+
+    db.session.commit()
+
+    return {
+        "id": goal.id,
+        "task_ids": task_ids
+    }, 200
+
+def create_new_task_for_goal(goal, task_data):
+    task_data["goal_id"] = goal.id
+
+    new_task = Task.from_dict(task_data)
+    db.session.add(new_task)
+    db.session.commit
+
+    return new_task.to_dict(), 200
+
+@bp.post("", strict_slashes = False)
+def create_goal():
+    return create_model(Goal, request.get_json())
+
+@bp.post("/<goal_id>/tasks", strict_slashes = False)
+def create_task_with_goal(goal_id):
+    goal = validate_model(Goal, goal_id)
+    request_body = request.get_json()
+
+    if "task_ids" in request_body:
+        task_ids = request_body["task_ids"]
+        return associate_tasks_with_goal(goal, task_ids)
+
+    else:
+        return create_new_task_for_goal
+
+
+@bp.get("", strict_slashes = False)
+def get_all_goals():
+    return get_all_items(Goal)
+
+@bp.get("/<goal_id>/tasks")
+def get_tasks_by_goal(goal_id):
+    goal = validate_model(Goal, goal_id)
+
+    return{
+        "id": goal.id,
+        "title": goal.title,
+        "tasks": [task.to_dict() for task in goal.tasks]
+    }
+
+@bp.get("/<goal_id>")
+def get_one_goal(goal_id):
+    return get_one_item(Goal, goal_id)
+
+@bp.put("/<goal_id>", strict_slashes = False)
+def update_entire_goal(goal_id):
+    updated_goal = update_entire_item(Goal, goal_id, goal_update=True)
+    return updated_goal.to_dict(), 200
+
+@bp.patch("/<goal_id>", strict_slashes = False)
+def update_partial_goal(goal_id):
+    return update_partial_item(Goal, goal_id)
+
+@bp.patch("/<goal_id>/mark_complete", strict_slashes = False)
+def mark_complete(goal_id):
+    return mark_item_complete(Goal, goal_id)
+
+@bp.patch("/<goal_id>/mark_incomplete", strict_slashes = False)
+def mark_incomplete(goal_id):
+    return mark_item_incomplete(Goal, goal_id)
+
+@bp.delete("/<goal_id>", strict_slashes = False)
+def delete_goal(goal_id):
+    return delete_item(Goal, goal_id)

app/routes/route_utilities.py

@@ -0,0 +1,138 @@
+from flask import abort, make_response, request, Response
+from ..db import db
+from datetime import datetime
+import os
+import requests
+
+def validate_model(cls, id):
+    try:
+        id = int(id)
+    except ValueError:
+        response = {"message": f"{cls.__name__} id {id} is invalid"}
+        abort(make_response(response, 400))
+
+    query = db.select(cls).where(cls.id == id)
+    item = db.session.scalar(query)
+
+    if not item:
+        response = {"message": f"{cls.__name__} {id} not found"}
+        abort(make_response(response, 404))
+
+    return item
+
+
+def send_slack_notification(item_title):
+    slack_token = os.environ.get("SLACK_TOKEN")
+    if slack_token:
+        try:
+            requests.post(
+            "https://slack.com/api/chat.postMessage",
+            headers={"Authorization": f"Bearer {slack_token}"},
+            json={
+                "channel": "slack-api-testing",
+                "text": f"Someone just completed {item_title}"
+                }
+            )
+        except Exception as e:
+            print(f"Slack notification failed: {e}")
+
+
+'''none of these probably go here, but it seemed easier at the time...
+& while my code is solid, my brain is delicate & easy to break.'''           
+def create_model(cls, request_body):
+    try:
+        new_item = cls.from_dict(request_body)
+
+    except KeyError:
+        response = {"details": "Invalid data"}
+        abort(make_response(response, 400))
+
+    db.session.add(new_item)
+    db.session.commit()
+
+    return new_item.to_dict(), 201
+
+def get_all_items(cls):
+    query = db.select(cls)
+
+    title_param = request.args.get('title')
+    if title_param:
+        query = query.where(cls.title.ilike(f"{title_param}%"))
+
+    sort_param = request.args.get('sort', 'asc')
+
+    if sort_param == "desc":
+        query = query.order_by(cls.title.desc())
+    else:
+        query = query.order_by(cls.title.asc())
+
+    items = db.session.scalars(query)
+
+    items_response = []
+    for item in items:
+        items_response.append(item.to_dict())
+    return items_response
+
+def get_one_item(cls, id):
+    item = validate_model(cls, id)
+
+    return item.to_dict()
+
+def update_entire_item(cls, id, goal_update=False):
+    item = validate_model(cls, id)
+    request_body = request.get_json()
+
+    item.title = request_body["title"]
+    if not goal_update:
+        item.description = request_body["description"]
+    db.session.commit()
+
+    return item
+
+def update_partial_item(cls, id):
+    item = validate_model(cls, id)
+    request_body = request.get_json()
+
+    if 'title' in request_body:
+        item.title = request_body['title']
+
+    if 'description' in request_body:
+        item.description = request_body['description']
+
+    if 'is_complete' in request_body:
+        if request_body['is_complete']:
+            item.completed_at = datetime.now()
+        else:
+            item.completed_at = None
+
+    db.session.commit()
+
+    return Response(status=204, mimetype="application/json")
+
+def mark_item_complete(cls, id):
+    item = validate_model(cls, id)
+
+    item.completed_at = datetime.now()
+
+    db.session.commit()
+
+    send_slack_notification(item.title)
+
+    return Response(status=204, mimetype="application/json")
+
+def mark_item_incomplete(cls, id):
+    item = validate_model(cls, id)
+
+    item.completed_at = None
+
+    db.session.commit()
+
+    return Response(status=204, mimetype="application/json")
+
+def delete_item(cls, id):
+    item = validate_model(cls, id)
+
+    db.session.delete(item)
+    db.session.commit()
+
+    return Response(status=204, mimetype="application/json")

app/routes/task_routes.py

@@ -1 +1,44 @@
-from flask import Blueprint
+from flask import Blueprint
+from app.models.task import Task
+from .route_utilities import *
+
+
+bp = Blueprint("tasks_bp", __name__, url_prefix="/tasks")
+
+@bp.post("", strict_slashes = False)
+def create_task():
+    return create_model(Task, request.get_json())
+
+@bp.get("", strict_slashes = False)
+def get_all_tasks():
+    return get_all_items(Task)
+
+@bp.get("/<task_id>")
+def get_one_task(task_id):
+    return get_one_item(Task, task_id)
+
+@bp.put("/<task_id>", strict_slashes = False)
+def update_entire_task(task_id):
+    update_entire_item(Task, task_id)
+    return Response(status=204, mimetype="application/json")
+
+@bp.patch("/<task_id>", strict_slashes = False)
+def update_partial_task(task_id):
+    return update_partial_item(Task, task_id)
+
+@bp.patch("/<task_id>/mark_complete", strict_slashes = False)
+def mark_complete(task_id):
+    return mark_item_complete(Task, task_id)
+
+@bp.patch("/<task_id>/mark_incomplete", strict_slashes = False)
+def mark_incomplete(task_id):
+    return mark_item_incomplete(Task, task_id)
+
+@bp.delete("/<task_id>", strict_slashes = False)
+def delete_task(task_id):
+    return delete_item(Task, task_id)
+
+
+
+
+

migrations/README

@@ -0,0 +1 @@
+Single-database configuration for Flask.