# Session 2 – Full Backend CRUD with Flask + MySQL + SQLAlchemy (ORM)

> **Goal:** build a production-style **backend-only** Task Manager API (no frontend) using **Flask**, **MySQL**, and **SQLAlchemy** with clear explanations for every key line. Everything below is ready to copy into files.

---

## 🧭 What You’ll Build
A REST API that supports:
- `GET /tasks` – list tasks
- `POST /tasks` – create task
- `PUT /tasks/<id>` – update task
- `DELETE /tasks/<id>` – delete task

Data is stored in a **MySQL** database via **SQLAlchemy** ORM models.

---

## ⚙️ Environment Setup (Anaconda)

# 1) Create & activate environment
conda create -n flask_crud python=3.11 -y
conda activate flask_crud

# 2) Install dependencies
pip install flask flask_sqlalchemy pymysql python-dotenv

# (Optional) useful during dev:
pip install ipython

**Why these packages?**
- `flask` – the web framework
- `flask_sqlalchemy` – ORM integration
- `pymysql` – pure-Python MySQL driver (easier to install than mysqlclient)
- `python-dotenv` – load DB credentials from a `.env` file instead of hardcoding

---

## 🗂 Project Structure

flask_task_manager/
│── app.py            # Flask app & routes (endpoints)
│── config.py         # Loads environment variables & builds DB URI
│── models.py         # SQLAlchemy models (Task)
│── db_setup.py       # One-time DB/table initialization script
│── .env              # Local secrets (DB credentials)
│── requirements.txt  # (optional) pinned deps for reproducible installs

> You’ll create each file using the code blocks below.

---

## 🔐 1) `.env` – Local Credentials (never commit to Git)

# .env
DB_USER=root
DB_PASSWORD=your_mysql_password
DB_HOST=127.0.0.1
DB_PORT=3306
DB_NAME=task_db
FLASK_ENV=development
SECRET_KEY=change-this-in-production

**Explanation**
- We avoid hardcoding secrets in code. `python-dotenv` will load these values into environment variables so `config.py` can read them.
- Replace `your_mysql_password` with your real MySQL password.
- Make sure the database **exists** (see next step).

---

## 🗄 2) Create the MySQL Database

Open your MySQL client and run:

CREATE DATABASE IF NOT EXISTS task_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

**Explanation**
- Creates a database named `task_db` if it doesn’t exist.
- `utf8mb4` supports full Unicode (including emojis).

---

## ⚙️ 3) `config.py` – Central Configuration

# config.py
import os
from dotenv import load_dotenv

# Load variables from .env into process environment
load_dotenv()

# Read values with defaults for safety
DB_USER = os.getenv("DB_USER", "root")
DB_PASSWORD = os.getenv("DB_PASSWORD", "")
DB_HOST = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT = os.getenv("DB_PORT", "3306")
DB_NAME = os.getenv("DB_NAME", "task_db")
SECRET_KEY = os.getenv("SECRET_KEY", "dev-secret")

# Build a SQLAlchemy-compatible connection URI using PyMySQL driver
SQLALCHEMY_DATABASE_URI = (
    f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
)

# Disable event system overhead we don’t use
SQLALCHEMY_TRACK_MODIFICATIONS = False

**Explanation (line-by-line highlights)**
- `load_dotenv()` reads `.env` so we can access env vars via `os.getenv`.
- `SQLALCHEMY_DATABASE_URI` uses the **PyMySQL** driver (`mysql+pymysql://…`).
- `SECRET_KEY` is needed by Flask for securely signing sessions (even if you don’t use sessions now).

---

## 🧱 4) `models.py` – SQLAlchemy Models

# models.py
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime

db = SQLAlchemy()

class Task(db.Model):
    __tablename__ = "tasks"

    id = db.Column(db.Integer, primary_key=True)
    content = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, default=False, nullable=False)
    created_at = db.Column(db.DateTime, default=datetime.utcnow, nullable=False)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False)

    def to_dict(self):
        return {
            "id": self.id,
            "content": self.content,
            "done": self.done,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
        }

**Explanation**
- `db = SQLAlchemy()` – global DB object the app will initialize.
- `Task` represents a table named `tasks`.
- `content` cannot be null (`nullable=False`). This is a **business rule** enforced at DB level.
- `created_at` and `updated_at` auto-manage timestamps:
- `default=datetime.utcnow` when the row is inserted
- `onupdate=datetime.utcnow` updates timestamp on any change
- `.to_dict()` serializes the model into a JSON-friendly shape for API responses.

---

## 🧰 5) `app.py` – Flask App & REST Endpoints

# app.py
from flask import Flask, request, jsonify
from models import db, Task
import config

def create_app():
    app = Flask(__name__)

    # Load configuration
    app.config["SQLALCHEMY_DATABASE_URI"] = config.SQLALCHEMY_DATABASE_URI
    app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = config.SQLALCHEMY_TRACK_MODIFICATIONS
    app.config["SECRET_KEY"] = config.SECRET_KEY

    # Initialize database with app context
    db.init_app(app)

    # ---------- Health & Root ----------
    @app.route("/")
    def root():
        return jsonify({"message": "Task Manager API (Flask + MySQL + SQLAlchemy)"}), 200

    @app.route("/healthz")
    def health():
        # Lightweight health check
        return jsonify({"status": "ok"}), 200

    # ---------- CRUD: Tasks ----------
    @app.route("/tasks", methods=["GET"])
    def list_tasks():
        """List all tasks."""
        tasks = Task.query.order_by(Task.created_at.desc()).all()
        return jsonify([t.to_dict() for t in tasks]), 200

    @app.route("/tasks/<int:task_id>", methods=["GET"])
    def get_task(task_id):
        """Get a single task by id."""
        task = Task.query.get(task_id)
        if not task:
            return jsonify({"error": "Task not found"}), 404
        return jsonify(task.to_dict()), 200

    @app.route("/tasks", methods=["POST"])
    def create_task():
        """Create a new task."""
        payload = request.get_json(silent=True) or {}
        content = payload.get("content", "").strip()

        if not content:
            return jsonify({"error": "Field 'content' is required and cannot be empty."}), 400

        task = Task(content=content, done=bool(payload.get("done", False)))
        db.session.add(task)
        db.session.commit()
        return jsonify(task.to_dict()), 201

    @app.route("/tasks/<int:task_id>", methods=["PUT"])
    def update_task(task_id):
        """Update content/done for an existing task."""
        task = Task.query.get(task_id)
        if not task:
            return jsonify({"error": "Task not found"}), 404

        payload = request.get_json(silent=True) or {}

        # Only update provided fields
        if "content" in payload:
            new_content = str(payload["content"]).strip()
            if not new_content:
                return jsonify({"error": "Field 'content' cannot be empty."}), 400
            task.content = new_content

        if "done" in payload:
            task.done = bool(payload["done"])

        db.session.commit()
        return jsonify(task.to_dict()), 200

    @app.route("/tasks/<int:task_id>", methods=["DELETE"])
    def delete_task(task_id):
        """Delete a task by id."""
        task = Task.query.get(task_id)
        if not task:
            return jsonify({"error": "Task not found"}), 404
        db.session.delete(task)
        db.session.commit()
        return jsonify({"message": "Task deleted"}), 200

    # ---------- Convenience Filters ----------
    @app.route("/tasks/done", methods=["GET"])
    def list_done():
        tasks = Task.query.filter_by(done=True).order_by(Task.updated_at.desc()).all()
        return jsonify([t.to_dict() for t in tasks]), 200

    @app.route("/tasks/pending", methods=["GET"])
    def list_pending():
        tasks = Task.query.filter_by(done=False).order_by(Task.created_at.desc()).all()
        return jsonify([t.to_dict() for t in tasks]), 200

    return app

# Dev entrypoint
if __name__ == "__main__":
    app = create_app()
    app.run(debug=True)

**Explanation (highlights)**
- `create_app()` pattern makes the app **factory-based** (helps testing and deployment).
- `db.init_app(app)` binds SQLAlchemy to this Flask instance.
- Each endpoint returns JSON + an appropriate **HTTP status code**.
- Validation rules:
- `POST /tasks` requires non-empty `content`.
- `PUT /tasks/<id>` only updates fields provided (partial update).
- Filter endpoints demonstrate ORM **querying** with `filter_by(...)` and `order_by(...)`.

---

## 🧪 6) `db_setup.py` – Initialize Tables

# db_setup.py
from app import create_app
from models import db

app = create_app()

with app.app_context():
    db.create_all()
    print("✅ Database tables created (or already exist).")

**Explanation**
- Imports the Flask app factory, enters an app context, then calls `db.create_all()` to create tables defined in `models.py`.

> Run it once after configuring `.env` and creating the database:
>
> ```bash
> python db_setup.py
> ```

---

## 📦 7) (Optional) `requirements.txt`

flask==3.0.3
flask_sqlalchemy==3.1.1
pymysql==1.1.1
python-dotenv==1.0.1

**Explanation**
- Helps pin versions so teammates install the same dependencies:

  pip install -r requirements.txt

---

## ▶️ Run the API

# Ensure DB exists and tables are created
python db_setup.py

# Start the API
python app.py

The server starts at `http://127.0.0.1:5000`.

---

## 🧪 Test with cURL

### Create a task

curl -X POST http://127.0.0.1:5000/tasks   -H "Content-Type: application/json"   -d '{"content": "Write Flask + MySQL API"}'

### List all tasks

curl http://127.0.0.1:5000/tasks

### Get a single task

curl http://127.0.0.1:5000/tasks/1

### Update a task

curl -X PUT http://127.0.0.1:5000/tasks/1   -H "Content-Type: application/json"   -d '{"done": true, "content":"Write Flask + MySQL API (updated)"}'

### Delete a task

curl -X DELETE http://127.0.0.1:5000/tasks/1

### List done / pending

curl http://127.0.0.1:5000/tasks/done
curl http://127.0.0.1:5000/tasks/pending

---

## 🧠 How ORM Maps to SQL (Conceptual)

- **Create**

  task = Task(content="Example")
  db.session.add(task)
  db.session.commit()

Behind the scenes SQL:

  INSERT INTO tasks (content, done, created_at, updated_at)
  VALUES ('Example', 0, NOW(), NOW());

- **Read**

  Task.query.filter_by(done=False).all()

Rough SQL:

  SELECT * FROM tasks WHERE done = 0;

- **Update**

  task.done = True
  db.session.commit()

Rough SQL:

  UPDATE tasks SET done = 1, updated_at = NOW() WHERE id = ?;

- **Delete**

  db.session.delete(task)
  db.session.commit()

Rough SQL:

  DELETE FROM tasks WHERE id = ?;

---

## 🛟 Troubleshooting

- **`ModuleNotFoundError: No module named 'pymysql'`**
Run `pip install pymysql` inside the `flask_crud` environment.

- **`(pymysql.err.OperationalError) (1045, "Access denied…")`**
Wrong credentials. Check `.env` values (user/password/host/port).

- **`Can't connect to MySQL server`**
Ensure MySQL is running and listening on `DB_HOST:DB_PORT`. On local dev, that’s usually `127.0.0.1:3306`.

- **`db.create_all()` but no tables**
Confirm you’re connecting to the *expected* database (`DB_NAME`). Print `config.SQLALCHEMY_DATABASE_URI` to verify.

- **Unicode issues**
Ensure DB uses `utf8mb4` character set (the `CREATE DATABASE` above does).

---

## 🧪 Stretch Goals (Homework)

1. Add **pagination** to `GET /tasks` using `?page=1&limit=20`.
2. Add **search**: `GET /tasks?query=<text>` filters by content.
3. Add **soft delete** (a `deleted_at` column instead of hard delete).
4. Add **SQLAlchemy migrations** (e.g., with Flask-Migrate) when your schema evolves.

---

Happy building! 🚀