Leveling Up My Task Automation API: Models, Routes, and Scheduling

Hey DEV community, I’m excited to share some fresh updates I pushed to my Task Automation API in this commit. This one’s all about getting the task model, routes, and scheduling logic in shape, setting the stage for a solid task management system. If you liked my HTTP status codes post, you’ll dig this dive into building a practical API. Let’s break it down.

What’s New in This Commit?

This commit (114bbb8) lays down the foundation for managing tasks. Here’s the quick rundown:

• Task Model: Upgraded it with clean, reliable fields.
• Routes: Added endpoints to schedule, list, and cancel tasks.
• Scheduling Logic: Got tasks saving to the database smoothly.
• Task Types: Set up support for reminders and file cleanup.
• Validation: Used Pydantic to keep scheduling times on point.

Let’s take a closer look at what’s cooking.

---

1. Task Model Upgrade (app/models/tasks.py)

I gave the TaskModel in app/models/tasks.py a serious tune-up to make it robust and ready for action. Here’s what changed:

• UUID for IDs: Switched to uuid.uuid4() for unique task IDs to avoid any overlap.
• Better Fields: Refined user_id, task_type, schedule_time, and status with SQLAlchemy, adding constraints like nullable=False for required fields.
• Timezone Support: Added a utcnow() function to ensure schedule_time uses UTC, keeping things consistent across timezones.

Here’s the updated code:

import uuid
from sqlalchemy import Column, String, ForeignKey, DateTime
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime, timezone

Base = declarative_base()

def utcnow():
    return datetime.now(timezone.utc)

class TaskModel(Base):
    __tablename__ = "Tasks"
    id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
    user_id = Column(String, ForeignKey("users.id"), nullable=False)
    task_type = Column(String, nullable=False)
    schedule_time = Column(DateTime(timezone=True), nullable=False, default=utcnow)
    status = Column(String, nullable=False, default="scheduled")

This model is now set to handle tasks.

---

2. New Routes for Task Management (app/routers/tasks.py)

I added some key endpoints in app/routers/tasks.py to make the API functional:

1. Schedule a Task (POST /schedule): Lets users create and schedule tasks, validated with Pydantic’s TaskCreate schema.
2. List Tasks (GET /list_files): Pulls up all tasks for the logged-in user. (Heads-up: list_files is a naming oops—I’ll rename it to list_tasks soon.)
3. Cancel a Task (GET /cancel/{task_id}): Updates a task’s status to “cancelled” to ditch it.

Here’s the code:

from fastapi import APIRouter, Depends, HTTPException, status
from app.routers import get_current_user, get_db, TaskModel, Session, User
from app.schemas.Tasks import TaskCreate, TaskResponse, TaskStatus, TaskType
from app.utils.task import schedule_task

router = APIRouter()

@router.post("/schedule")
def schedule_logic(
    task: TaskCreate,
    db: Session = Depends(get_db),
    user: dict = Depends(get_current_user)
):
    try:
        new_task = schedule_task(db=db, user_id=user.id, task_data=task)
        return TaskResponse.model_validate(new_task)
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@router.get("/list_files", response_model=list[TaskResponse])
def list_tasks(
    db: Session = Depends(get_db),
    user: dict = Depends(get_current_user)
):
    tasks = db.query(TaskModel).filter(TaskModel.user_id == user['id']).all()
    return tasks

@router.get("/cancel/{task_id}", response_model=TaskResponse)
def cancel_task(
    task_id: int,
    db: Session = Depends(get_db),
    user: dict = Depends(get_current_user)
):
    task = db.query(TaskModel).filter(TaskModel.id == task_id, TaskModel.user_id == user['id']).first()
    if not task:
        raise HTTPException(status_code=400, detail="task doesnt exist or cant be found")
    task.status = "cancelled"
    db.commit()
    db.refresh(task)
    return task

These routes are locked down with authentication and handle database operations cleanly. I’m also using HTTP status codes (like 400 for errors), tying back to my HTTP status codes post for clear error communication.

---

3. Scheduling Logic and Task Types

The commit includes scheduling logic (in app/utils/task.py, not shown in the diff) that saves tasks to the database. I also added support for two task types:

• Reminders: For notifying users at a specific time.
• File Cleanup: For deleting files (still in progress).

Pydantic validation ensures scheduling times make sense, so no one’s trying to schedule a task for last week.

---

4. What’s Next?

This commit gets the API in a good spot, but there’s more to do:

• Task Execution: Need to implement the actual logic for reminders and file deletion.
• Fix Naming: list_files should be list_tasks—my bad.
• Better Errors: Plan to make error messages more specific.
• Scheduler Integration: Looking at tools like APScheduler or Celery to run tasks on schedule.

---

Why This Matters

This commit is a big step toward making the Task Automation API a practical tool for managing tasks, whether it’s reminders or cleaning up files. It’s built with FastAPI and SQLAlchemy, keeping things clean and scalable. If you’re into APIs or automation, this might spark some ideas for your own projects.

Check out the full commit here. Got thoughts or suggestions? Drop them in the comments—I’d love to hear what you think!

Happy coding!!