# Create project + install
python -m venv .venv
source .venv/bin/activate
pip install flask

from flask import Flask

app = Flask(__name__)        # __name__ = current module; Flask uses it
                             # to locate templates/ and static/

@app.route("/")
def home():
    return "<h1>Hello, World!</h1>"     # returning a string = HTML response

@app.route("/api/health")
def health():
    return {"status": "ok"}              # returning a dict = auto JSON!

@app.route("/about")
def about():
    return "This is the about page."

if __name__ == "__main__":
    app.run(debug=True)      # dev server with auto-reload + debugger

# Option 1 — run through the flask CLI (recommended)
flask --app main run --debug

# Option 2 — run the file directly
python main.py

# main     = the file main.py
# --debug  = auto-restart on changes + interactive debugger (DEV ONLY)

# Now visit:
# http://127.0.0.1:5000            → Hello, World!
# http://127.0.0.1:5000/api/health → {"status": "ok"}

from flask import Flask, request

app = Flask(__name__)

# ── Classic style: one decorator, methods list ──
@app.route("/users", methods=["GET", "POST"])
def users():
    if request.method == "POST":
        return {"created": True}, 201     # handle create
    return {"users": ["a", "b"]}          # handle list

# ── Modern shortcuts (Flask 2.0+) — one function per verb ──
@app.get("/items")
def list_items():
    return {"items": []}

@app.post("/items")
def create_item():
    return {"created": True}, 201

@app.put("/items/<int:item_id>")
def replace_item(item_id):
    return {"replaced": item_id}

@app.patch("/items/<int:item_id>")
def update_item(item_id):
    return {"patched": item_id}

@app.delete("/items/<int:item_id>")
def delete_item(item_id):
    return "", 204

# ── Multiple URLs → one view ──
@app.route("/")
@app.route("/home")
def home():
    return "Home page"

# ── url_for: build URLs from endpoint names ──
from flask import url_for, redirect

@app.get("/old-page")
def old_page():
    return redirect(url_for("home"))       # → "/"
    # url_for("list_items")                → "/items"
    # url_for("replace_item", item_id=7)   → "/items/7"
    # url_for("home", _external=True)      → "http://localhost:5000/"

# ── Basic capture (default = string) ──
@app.get("/user/<username>")
def show_user(username):
    return f"User: {username}"

# GET /user/yatin → "User: yatin"

# ── Typed converters ──
@app.get("/post/<int:post_id>")
def show_post(post_id):
    return {"post_id": post_id, "type": str(type(post_id).__name__)}

# GET /post/42  → {"post_id": 42, "type": "int"}
# GET /post/abc → 404 Not Found  (route never matches)

@app.get("/files/<path:subpath>")
def show_file(subpath):
    return f"Path: {subpath}"        # /files/docs/a/b.txt → "docs/a/b.txt"

@app.get("/order/<uuid:order_id>")
def show_order(order_id):
    return {"order": str(order_id)}  # order_id is a real uuid.UUID object

# ── Multiple variables ──
@app.get("/repos/<owner>/<repo>/issues/<int:number>")
def issue(owner, repo, number):
    return {"owner": owner, "repo": repo, "issue": number}

# ── Fixed choices ──
@app.get("/<any(about, contact, help):page>")
def static_page(page):
    return f"Rendering the {page} page"

# ── Custom converter (advanced) ──
from werkzeug.routing import BaseConverter

class SlugConverter(BaseConverter):
    regex = r"[a-z0-9]+(?:-[a-z0-9]+)*"     # lowercase-words-with-dashes

app.url_map.converters["slug"] = SlugConverter

@app.get("/blog/<slug:post_slug>")
def blog_post(post_slug):
    return f"Post: {post_slug}"

from flask import Flask, request

app = Flask(__name__)

# ── Query parameters:  GET /search?q=laptop&page=2&tag=new&tag=sale ──
@app.get("/search")
def search():
    q     = request.args.get("q", default="", type=str)
    page  = request.args.get("page", default=1, type=int)   # "2" → 2
    limit = request.args.get("limit", default=10, type=int)
    tags  = request.args.getlist("tag")                     # ["new", "sale"]

    if not q:
        return {"error": "q is required"}, 400              # manual validation!

    return {"q": q, "page": page, "limit": limit, "tags": tags}

# ── Form data:  POST with <form method="post"> ──
@app.post("/register")
def register():
    name  = request.form.get("name", "").strip()
    email = request.form["email"]        # missing key → automatic 400 response
    age   = request.form.get("age", type=int)

    errors = []
    if len(name) < 2:
        errors.append("name too short")
    if age is not None and not (0 <= age <= 150):
        errors.append("age out of range")
    if errors:
        return {"errors": errors}, 400

    return {"name": name, "email": email, "age": age}, 201

# ── JSON body:  POST /api/users  {"name": "Yatin", "email": "y@dev.com"} ──
@app.post("/api/users")
def create_user():
    data = request.get_json(silent=True)   # None if body isn't valid JSON
    if data is None:
        return {"error": "expected JSON body"}, 400

    name  = data.get("name")
    email = data.get("email")
    if not name or not email:
        return {"error": "name and email are required"}, 400

    return {"created": {"name": name, "email": email}}, 201

from flask import Flask, request

app = Flask(__name__)

@app.route("/inspect", methods=["GET", "POST"])
def inspect():
    return {
        "method":       request.method,
        "path":         request.path,
        "url":          request.url,
        "query":        request.args.to_dict(),
        "headers": {
            "user_agent":  request.headers.get("User-Agent"),
            "auth":        request.headers.get("Authorization"),
            "content_type": request.content_type,
        },
        "cookies":      request.cookies.to_dict() if request.cookies else {},
        "client_ip":    request.remote_addr,
        "is_json":      request.is_json,
        "endpoint":     request.endpoint,     # "inspect"
        "view_args":    request.view_args,     # matched URL variables
    }

# ── Using the request in helper functions (no parameter passing!) ──
def current_locale():
    # request works anywhere *during* a request — even in helpers
    return request.headers.get("Accept-Language", "en")[:2]

@app.get("/greet")
def greet():
    lang = current_locale()
    return {"hello": "hola" if lang == "es" else "hello"}

from flask import Flask, jsonify, make_response, redirect, url_for, abort, send_file, Response, stream_with_context

app = Flask(__name__)

# ── The five common shapes ──
@app.get("/text")
def text():
    return "<h1>Plain HTML string</h1>"                  # 200 text/html

@app.get("/json")
def as_json():
    return {"name": "Yatin", "role": "dev"}               # 200 application/json

@app.post("/created")
def created():
    return {"id": 7}, 201                                  # status code

@app.get("/with-headers")
def with_headers():
    return {"data": []}, 200, {"X-Total-Count": "0",
                               "Cache-Control": "no-store"}

@app.get("/old")
def old():
    return redirect(url_for("as_json"))                   # 302 → /json

# ── make_response: when you need to tweak before sending ──
@app.get("/custom")
def custom():
    resp = make_response({"ok": True}, 200)
    resp.headers["X-Request-Id"] = "abc-123"
    resp.set_cookie("seen", "1", httponly=True)
    return resp

# ── abort: bail out with an HTTP error ──
FAKE_DB = {1: {"name": "Yatin"}}

@app.get("/users/<int:user_id>")
def get_user(user_id):
    user = FAKE_DB.get(user_id)
    if user is None:
        abort(404, description="User not found")          # → error handler
    return user

# ── Files ──
@app.get("/invoice")
def invoice():
    return send_file("files/invoice.pdf",
                     as_attachment=True,                   # force download
                     download_name="invoice-2026.pdf")

# ── Streaming large responses (CSV export, logs, SSE) ──
@app.get("/export")
def export():
    def generate():
        yield "id,name\n"
        for i in range(100_000):
            yield f"{i},user{i}\n"                        # sent chunk by chunk
    return Response(stream_with_context(generate()),
                    mimetype="text/csv")

from flask import Flask, render_template

app = Flask(__name__)

@app.get("/profile/<username>")
def profile(username):
    user = {"name": username, "role": "admin", "score": 92}
    posts = [
        {"title": "Learning Flask", "likes": 14},
        {"title": "Jinja is neat",  "likes": 31},
    ]
    return render_template("profile.html", user=user, posts=posts)

# ── Custom filter available in ALL templates ──
@app.template_filter("shout")
def shout(text):
    return str(text).upper() + "!"

# ── Context processor: inject variables into every template ──
@app.context_processor
def inject_globals():
    return {"site_name": "Yatin Notes", "year": 2026}

<!DOCTYPE html>
<html>
<head>
  <title>{% block title %}{{ site_name }}{% endblock %}</title>
  <link rel="stylesheet" href="{{ url_for('static', filename='style.css') }}">
</head>
<body>
  <nav>
    <a href="{{ url_for('profile', username='yatin') }}">Profile</a>
  </nav>

  {% block content %}{% endblock %}      {# pages fill this in #}

  <footer>© {{ year }} {{ site_name }}</footer>
</body>
</html>

{% extends "base.html" %}

{% block title %}{{ user.name }} — {{ site_name }}{% endblock %}

{% block content %}
  <h1>{{ user.name|title }} <small>({{ user.role }})</small></h1>

  {% if user.score > 90 %}
    <p>{{ "top performer"|shout }}</p>      {# custom filter #}
  {% elif user.score > 50 %}
    <p>Doing well.</p>
  {% else %}
    <p>Keep going!</p>
  {% endif %}

  <ul>
  {% for post in posts %}
    <li>{{ loop.index }}. {{ post.title }} — {{ post.likes }} likes</li>
  {% else %}
    <li>No posts yet.</li>                 {# runs when the list is empty #}
  {% endfor %}
  </ul>
{% endblock %}

myapp/
├── app.py
├── static/
│   ├── css/style.css
│   ├── js/app.js
│   └── img/logo.png
└── templates/
    └── index.html

<link rel="stylesheet"
      href="{{ url_for('static', filename='css/style.css') }}">
<script src="{{ url_for('static', filename='js/app.js') }}" defer></script>
<img src="{{ url_for('static', filename='img/logo.png') }}" alt="logo">

# Let nginx serve static files directly — Python never sees them
location /static/ {
    alias /srv/myapp/static/;
    expires 30d;
    add_header Cache-Control "public, immutable";
}

from flask import Flask, abort, jsonify, render_template, request
from werkzeug.exceptions import HTTPException

app = Flask(__name__)

# ── abort() from anywhere ──
USERS = {1: "Yatin"}

@app.get("/users/<int:user_id>")
def get_user(user_id):
    if user_id not in USERS:
        abort(404, description=f"User {user_id} not found")
    return {"id": user_id, "name": USERS[user_id]}

# ── Custom HTML error pages (for websites) ──
@app.errorhandler(404)
def not_found(error):
    return render_template("errors/404.html", msg=error.description), 404

@app.errorhandler(500)
def server_error(error):
    return render_template("errors/500.html"), 500

# ── JSON errors for EVERY HTTP status (for APIs) ──
@app.errorhandler(HTTPException)
def handle_http_error(e):
    return jsonify({
        "error":  e.name,            # "Not Found"
        "detail": e.description,     # your abort() message
        "status": e.code,            # 404
    }), e.code

# ── Map domain exceptions → HTTP ──
class OutOfStock(Exception):
    def __init__(self, product_id):
        self.product_id = product_id

@app.errorhandler(OutOfStock)
def handle_out_of_stock(e):
    return {"error": "out_of_stock", "product": e.product_id}, 409

@app.post("/buy/<int:pid>")
def buy(pid):
    raise OutOfStock(pid)            # business code stays HTTP-free

# ── Last-resort 500 with logging ──
@app.errorhandler(Exception)
def handle_unexpected(e):
    if isinstance(e, HTTPException):     # let real HTTP errors pass through
        raise e
    app.logger.exception("Unhandled error on %s", request.path)
    return {"error": "internal_error"}, 500

from flask import Flask, request, make_response

app = Flask(__name__)

# ── Reading headers ──
@app.get("/whoami")
def whoami():
    return {
        "user_agent": request.headers.get("User-Agent"),
        "token":      request.headers.get("Authorization"),   # "Bearer xyz"
        "language":   request.headers.get("Accept-Language"),
        "custom":     request.headers.get("X-Client-Version", "unknown"),
    }

# ── Reading cookies ──
@app.get("/prefs")
def prefs():
    theme = request.cookies.get("theme", "light")
    return {"theme": theme}

# ── Setting headers + cookies ──
@app.post("/prefs")
def set_prefs():
    theme = request.get_json().get("theme", "light")

    resp = make_response({"saved": True})
    resp.set_cookie(
        "theme", theme,
        max_age=60 * 60 * 24 * 365,   # 1 year
        httponly=False,               # UI cookie — JS may read it
        secure=True,
        samesite="Lax",
    )
    resp.headers["X-App-Version"] = "3.2.1"
    return resp

# ── Auth-style cookie (locked down) ──
@app.post("/login")
def login():
    resp = make_response({"message": "logged in"})
    resp.set_cookie("auth_token", "signed-token-here",
                    httponly=True, secure=True, samesite="Lax",
                    max_age=60 * 30)
    return resp

@app.post("/logout")
def logout():
    resp = make_response({"message": "bye"})
    resp.delete_cookie("auth_token")
    return resp

from flask import Flask, session, redirect, url_for, request, flash

app = Flask(__name__)
app.secret_key = "change-me-load-from-env"      # signs the session cookie
# production: app.config["SECRET_KEY"] = os.environ["SECRET_KEY"]

# ── Login: put the user id in the session ──
@app.post("/login")
def login():
    username = request.form["username"]
    password = request.form["password"]

    user = verify_credentials(username, password)   # your DB check
    if not user:
        flash("Invalid credentials")
        return redirect(url_for("login_page"))

    session.clear()                     # prevent session fixation
    session["user_id"] = user.id
    session["role"]    = user.role
    session.permanent  = True           # survive browser restarts
    return redirect(url_for("dashboard"))

# ── Protect pages ──
@app.get("/dashboard")
def dashboard():
    if "user_id" not in session:
        return redirect(url_for("login_page"))
    return f"Welcome back, user #{session['user_id']} ({session['role']})"

# ── Logout ──
@app.post("/logout")
def logout():
    session.clear()
    return redirect(url_for("login_page"))

# ── Non-auth uses: preferences, wizards, carts (small!) ──
@app.post("/theme/<name>")
def set_theme(name):
    session["theme"] = name
    return {"theme": name}

# pip install flask-session redis
from flask_session import Session
import redis

app.config.update(
    SESSION_TYPE="redis",
    SESSION_REDIS=redis.from_url("redis://localhost:6379"),
    SESSION_PERMANENT=False,
)
Session(app)
# Now the cookie holds only a random ID; the data lives in Redis.
# → revocable logins, no 4KB limit, nothing readable client-side.

from flask import Blueprint, render_template, abort

# name, import location, options
bp = Blueprint("blog", __name__, url_prefix="/blog",
               template_folder="templates")

POSTS = {1: {"title": "Hello Flask", "body": "..."}}

@bp.get("/")                       # → GET /blog/
def index():
    return render_template("blog/index.html", posts=POSTS)

@bp.get("/<int:post_id>")          # → GET /blog/<id>
def show_post(post_id):
    post = POSTS.get(post_id) or abort(404)
    return render_template("blog/post.html", post=post)

# Hook scoped to THIS blueprint only
@bp.before_request
def count_blog_hits():
    ...   # runs before every /blog/* request

# Error page scoped to THIS blueprint only
@bp.errorhandler(404)
def blog_404(e):
    return render_template("blog/404.html"), 404

# api/routes.py — a versioned JSON API as a blueprint
from flask import Blueprint

api = Blueprint("api", __name__, url_prefix="/api/v1")

@api.get("/users")
def list_users():
    return {"users": []}

@api.get("/users/<int:uid>")
def get_user(uid):
    return {"id": uid}


# main.py — wire everything together
from flask import Flask
from blog.routes import bp as blog_bp
from api.routes import api as api_bp

app = Flask(__name__)
app.register_blueprint(blog_bp)          # /blog/...
app.register_blueprint(api_bp)           # /api/v1/...

# Same blueprint, second mount point (e.g., legacy path)
# app.register_blueprint(api_bp, url_prefix="/api/latest", name="api_latest")

# Nested blueprints (Flask 2.0+)
# parent = Blueprint("admin", __name__, url_prefix="/admin")
# parent.register_blueprint(reports_bp)   # → /admin/reports/...
# app.register_blueprint(parent)

from flask import Flask, g, current_app, request
import sqlite3, time, uuid

app = Flask(__name__)
app.config["DATABASE"] = "app.db"

# ── g: compute once per request, reuse everywhere ──
def get_db():
    if "db" not in g:                                # first call this request?
        g.db = sqlite3.connect(current_app.config["DATABASE"])
    return g.db                                      # later calls reuse it

@app.teardown_appcontext
def close_db(exc):
    db = g.pop("db", None)
    if db is not None:
        db.close()                                   # always runs, even on errors

@app.before_request
def stamp_request():
    g.request_id = uuid.uuid4().hex[:8]
    g.start_time = time.perf_counter()

@app.get("/report")
def report():
    rows = get_db().execute("SELECT count(*) FROM users").fetchone()
    return {"request_id": g.request_id, "users": rows[0]}

# ── current_app inside a blueprint (no circular import) ──
from flask import Blueprint
bp = Blueprint("tools", __name__)

@bp.get("/config-peek")
def config_peek():
    return {"debug": current_app.debug,
            "db": current_app.config["DATABASE"]}

# ── Using the app OUTSIDE a request (script / shell / cron) ──
def seed():
    with app.app_context():          # manually push the app context
        db = get_db()
        db.execute("INSERT INTO users(name) VALUES ('admin')")
        db.commit()

if __name__ == "__main__":
    seed()

from flask import Flask, request, g, abort
import time, logging

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)

# ── before_request: timing + a simple auth gate ──
PUBLIC_ENDPOINTS = {"health", "login", "static"}

@app.before_request
def start_timer():
    g.start = time.perf_counter()

@app.before_request
def require_token():
    if request.endpoint in PUBLIC_ENDPOINTS or request.endpoint is None:
        return                                   # allowed through
    token = request.headers.get("Authorization", "")
    if not token.startswith("Bearer "):
        abort(401, description="Missing bearer token")
        # returning/raising here SKIPS the view entirely

# ── after_request: security headers + timing header + access log ──
@app.after_request
def finalize(response):
    elapsed = (time.perf_counter() - g.get("start", time.perf_counter())) * 1000
    response.headers["X-Response-Time"] = f"{elapsed:.1f}ms"
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    app.logger.info("%s %s → %s (%.1fms)",
                    request.method, request.path, response.status_code, elapsed)
    return response                              # MUST return it

# ── teardown: guaranteed cleanup ──
@app.teardown_appcontext
def cleanup(exc):
    db = g.pop("db", None)
    if db is not None:
        db.close()

@app.get("/health")
def health():
    return {"status": "ok"}

# ── ProxyFix: REQUIRED when running behind nginx / a load balancer ──
# Makes request.remote_addr and url scheme reflect the real client,
# by trusting X-Forwarded-* headers from exactly N proxies.
from werkzeug.middleware.proxy_fix import ProxyFix

app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)

# ── Writing your own WSGI middleware (framework-agnostic) ──
class SimpleWSGILogger:
    def __init__(self, wsgi_app):
        self.wsgi_app = wsgi_app
    def __call__(self, environ, start_response):
        print("WSGI sees:", environ["REQUEST_METHOD"], environ["PATH_INFO"])
        return self.wsgi_app(environ, start_response)

app.wsgi_app = SimpleWSGILogger(app.wsgi_app)

# pip install flask-login flask-sqlalchemy
from flask import Flask, request, redirect, url_for, render_template, flash
from flask_login import (LoginManager, UserMixin, login_user,
                         logout_user, login_required, current_user)
from werkzeug.security import generate_password_hash, check_password_hash

app = Flask(__name__)
app.secret_key = "load-from-env"

login_manager = LoginManager(app)
login_manager.login_view = "login"        # where @login_required redirects

# ── User model (SQLAlchemy version in topic 17) ──
class User(UserMixin):                     # UserMixin adds is_authenticated etc.
    def __init__(self, id, email, pw_hash):
        self.id, self.email, self.pw_hash = id, email, pw_hash

USERS = {"1": User("1", "y@dev.com", generate_password_hash("secret123"))}

@login_manager.user_loader
def load_user(user_id):                    # called on EVERY request
    return USERS.get(user_id)              # id comes from the session cookie

# ── Routes ──
@app.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        user = next((u for u in USERS.values()
                     if u.email == request.form["email"]), None)
        if user and check_password_hash(user.pw_hash, request.form["password"]):
            login_user(user, remember=True)
            return redirect(url_for("dashboard"))
        flash("Invalid credentials")       # same message for both failures
    return render_template("login.html")

@app.get("/dashboard")
@login_required                            # anonymous → redirect to /login
def dashboard():
    return f"Hello {current_user.email}!"

@app.post("/logout")
@login_required
def logout():
    logout_user()
    return redirect(url_for("login"))

# pip install flask-jwt-extended
from flask import Flask, request
from flask_jwt_extended import (JWTManager, create_access_token,
                                create_refresh_token, jwt_required,
                                get_jwt_identity)
from werkzeug.security import check_password_hash

app = Flask(__name__)
app.config["JWT_SECRET_KEY"] = "load-from-env"
app.config["JWT_ACCESS_TOKEN_EXPIRES"] = 60 * 30        # 30 minutes
jwt = JWTManager(app)

@app.post("/api/login")
def api_login():
    data = request.get_json()
    user = find_user_by_email(data.get("email"))         # your DB lookup
    if not user or not check_password_hash(user.pw_hash, data.get("password", "")):
        return {"error": "invalid credentials"}, 401
    return {
        "access_token":  create_access_token(identity=str(user.id)),
        "refresh_token": create_refresh_token(identity=str(user.id)),
    }

@app.get("/api/me")
@jwt_required()                            # verifies Authorization: Bearer <jwt>
def me():
    user_id = get_jwt_identity()
    return {"user_id": user_id}

@app.post("/api/refresh")
@jwt_required(refresh=True)                # only refresh tokens allowed here
def refresh():
    return {"access_token": create_access_token(identity=get_jwt_identity())}

# Client usage:
#   POST /api/login {"email": ..., "password": ...}  → tokens
#   GET  /api/me    Authorization: Bearer <access_token>

# pip install flask-sqlalchemy
from flask import Flask, request, abort
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timezone

app = Flask(__name__)
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///app.db"

db = SQLAlchemy(app)     # (factory apps: db = SQLAlchemy() + db.init_app(app))

class User(db.Model):
    __tablename__ = "users"
    id        = db.Column(db.Integer, primary_key=True)
    name      = db.Column(db.String(100), nullable=False)
    email     = db.Column(db.String(255), unique=True, index=True, nullable=False)
    created_at = db.Column(db.DateTime,
                           default=lambda: datetime.now(timezone.utc))
    posts     = db.relationship("Post", back_populates="author",
                                cascade="all, delete-orphan")

class Post(db.Model):
    __tablename__ = "posts"
    id       = db.Column(db.Integer, primary_key=True)
    title    = db.Column(db.String(200), nullable=False)
    user_id  = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False)
    author   = db.relationship("User", back_populates="posts")

with app.app_context():
    db.create_all()          # dev convenience; real schema changes → migrations

# ── CREATE ──
@app.post("/users")
def create_user():
    data = request.get_json()
    if db.session.execute(
        db.select(User).filter_by(email=data["email"])
    ).scalar_one_or_none():
        return {"error": "email already registered"}, 409

    user = User(name=data["name"], email=data["email"])
    db.session.add(user)
    db.session.commit()
    return {"id": user.id, "name": user.name}, 201

# ── READ (list, paginated) ──
@app.get("/users")
def list_users():
    page = db.paginate(db.select(User).order_by(User.id),
                       page=request.args.get("page", 1, type=int),
                       per_page=10, error_out=False)
    return {
        "items": [{"id": u.id, "name": u.name} for u in page.items],
        "page": page.page, "pages": page.pages, "total": page.total,
    }

# ── READ (single, auto-404) ──
@app.get("/users/<int:user_id>")
def get_user(user_id):
    user = db.get_or_404(User, user_id)
    return {"id": user.id, "name": user.name, "email": user.email,
            "posts": [p.title for p in user.posts]}

# ── UPDATE (partial) ──
@app.patch("/users/<int:user_id>")
def update_user(user_id):
    user = db.get_or_404(User, user_id)
    data = request.get_json()
    if "name" in data:  user.name  = data["name"]
    if "email" in data: user.email = data["email"]
    db.session.commit()
    return {"id": user.id, "name": user.name}

# ── DELETE ──
@app.delete("/users/<int:user_id>")
def delete_user(user_id):
    user = db.get_or_404(User, user_id)
    db.session.delete(user)
    db.session.commit()
    return "", 204

# ── Query patterns cheat-sheet ──
# db.session.execute(db.select(User)).scalars().all()            # all
# db.session.execute(db.select(User).filter_by(name="Yatin")
#                   ).scalar_one_or_none()                        # one or None
# db.session.execute(db.select(User)
#     .where(User.email.like("%@dev.com"))
#     .order_by(User.created_at.desc()).limit(5)).scalars().all()
# Legacy style (still everywhere): User.query.filter_by(...).first()

# pip install flask-migrate
from flask_migrate import Migrate

migrate = Migrate(app, db)        # factory apps: migrate.init_app(app, db)

# One-time setup
flask db init

# 1) You edit models.py — e.g., add User.bio column
# 2) Generate the migration
flask db migrate -m "add bio to users"

# 3) REVIEW migrations/versions/xxxx_add_bio_to_users.py
#    def upgrade():
#        op.add_column('users', sa.Column('bio', sa.String(500)))
#    def downgrade():
#        op.drop_column('users', 'bio')

# 4) Apply it
flask db upgrade

# On the server, deploys just run:  flask db upgrade
# Oops?  flask db downgrade

# pip install flask-wtf email-validator
from flask import Flask, render_template, redirect, url_for, flash
from flask_wtf import FlaskForm
from wtforms import StringField, PasswordField, IntegerField, SubmitField
from wtforms.validators import (DataRequired, Length, Email,
                                EqualTo, NumberRange, ValidationError)

app = Flask(__name__)
app.secret_key = "load-from-env"          # powers CSRF tokens too

class RegisterForm(FlaskForm):
    name     = StringField("Name",
                 validators=[DataRequired(), Length(min=2, max=50)])
    email    = StringField("Email",
                 validators=[DataRequired(), Email()])
    age      = IntegerField("Age",
                 validators=[NumberRange(min=13, max=120)])
    password = PasswordField("Password",
                 validators=[DataRequired(), Length(min=8)])
    confirm  = PasswordField("Confirm",
                 validators=[EqualTo("password", message="Passwords must match")])
    submit   = SubmitField("Create account")

    def validate_email(self, field):      # custom rule: auto-runs for `email`
        if email_taken(field.data):        # your DB check
            raise ValidationError("That email is already registered.")

@app.route("/register", methods=["GET", "POST"])
def register():
    form = RegisterForm()
    if form.validate_on_submit():          # POST + all validators + CSRF ok
        create_user(form.name.data, form.email.data, form.password.data)
        flash("Welcome aboard!")
        return redirect(url_for("login"))
    return render_template("register.html", form=form)   # GET or errors

<form method="post" novalidate>
  {{ form.hidden_tag() }}                 {# ← the CSRF token. NEVER omit. #}

  {{ form.name.label }} {{ form.name(size=30) }}
  {% for err in form.name.errors %}<small class="err">{{ err }}</small>{% endfor %}

  {{ form.email.label }} {{ form.email() }}
  {% for err in form.email.errors %}<small class="err">{{ err }}</small>{% endfor %}

  {{ form.password.label }} {{ form.password() }}
  {{ form.confirm.label }} {{ form.confirm() }}

  {{ form.submit() }}
</form>

# pip install marshmallow
from marshmallow import Schema, fields, validate, ValidationError

class UserSchema(Schema):
    name  = fields.Str(required=True, validate=validate.Length(min=2, max=50))
    email = fields.Email(required=True)
    age   = fields.Int(load_default=None,
                       validate=validate.Range(min=0, max=150))

user_schema = UserSchema()

@app.post("/api/users")
def api_create_user():
    try:
        data = user_schema.load(request.get_json() or {})   # validate + coerce
    except ValidationError as err:
        return {"errors": err.messages}, 400   # {"email": ["Not a valid email."]}
    return {"created": data}, 201

import uuid
from pathlib import Path
from flask import Flask, request, abort, send_from_directory
from werkzeug.utils import secure_filename

app = Flask(__name__)
app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024      # 16 MB → else 413

UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
ALLOWED = {".png", ".jpg", ".jpeg", ".gif", ".pdf"}

def allowed(filename: str) -> bool:
    return Path(filename).suffix.lower() in ALLOWED

# ── Single file ──
@app.post("/upload")
def upload():
    if "file" not in request.files:
        return {"error": "no file part (did you set enctype?)"}, 400
    file = request.files["file"]
    if file.filename == "":
        return {"error": "no file selected"}, 400
    if not allowed(file.filename):
        return {"error": "file type not allowed"}, 400

    ext    = Path(secure_filename(file.filename)).suffix.lower()
    stored = f"{uuid.uuid4().hex}{ext}"       # unguessable, collision-proof
    file.save(UPLOAD_DIR / stored)

    return {"stored_as": stored,
            "original": file.filename,
            "content_type": file.content_type}, 201

# ── Multiple files:  <input type="file" name="files" multiple> ──
@app.post("/upload-many")
def upload_many():
    saved = []
    for file in request.files.getlist("files"):
        if file and allowed(file.filename):
            stored = f"{uuid.uuid4().hex}{Path(file.filename).suffix.lower()}"
            file.save(UPLOAD_DIR / stored)
            saved.append(stored)
    return {"saved": saved}, 201

# ── Serving uploads back (path-traversal safe) ──
@app.get("/files/<name>")
def get_file(name):
    return send_from_directory(UPLOAD_DIR, name)   # 404s outside the dir

# ── Friendly 413 ──
@app.errorhandler(413)
def too_big(e):
    return {"error": "file too large (max 16 MB)"}, 413

<form method="post" action="/upload" enctype="multipart/form-data">
  <input type="file" name="file">
  <button>Upload</button>
</form>

# pip install flask-cors
from flask import Flask, Blueprint, request, jsonify
from flask.views import MethodView
from flask_cors import CORS
from werkzeug.exceptions import HTTPException

app = Flask(__name__)

# ── CORS: allow the frontend origin(s) on /api/* ──
CORS(app,
     resources={r"/api/*": {"origins": ["http://localhost:3000",
                                        "https://myapp.com"]}},
     supports_credentials=True)

api = Blueprint("api", __name__, url_prefix="/api/v1")

USERS = {1: {"id": 1, "name": "Yatin", "email": "y@dev.com"}}
NEXT_ID = 2

# ── MethodView: one class per resource ──
class UserListAPI(MethodView):
    def get(self):
        return jsonify(list(USERS.values()))

    def post(self):
        global NEXT_ID
        data = request.get_json(silent=True) or {}
        if not data.get("name") or not data.get("email"):
            return {"error": "name and email required"}, 400
        user = {"id": NEXT_ID, **data}
        USERS[NEXT_ID] = user
        NEXT_ID += 1
        return user, 201

class UserAPI(MethodView):
    def get(self, user_id):
        user = USERS.get(user_id)
        return (user, 200) if user else ({"error": "not found"}, 404)

    def patch(self, user_id):
        user = USERS.get(user_id)
        if not user:
            return {"error": "not found"}, 404
        user.update(request.get_json(silent=True) or {})
        return user

    def delete(self, user_id):
        USERS.pop(user_id, None)
        return "", 204

api.add_url_rule("/users",
                 view_func=UserListAPI.as_view("user_list"))
api.add_url_rule("/users/<int:user_id>",
                 view_func=UserAPI.as_view("user_detail"))

# ── One JSON error shape for the whole API ──
@app.errorhandler(HTTPException)
def api_error(e):
    return {"error": {"status": e.code, "message": e.description or e.name}}, e.code

app.register_blueprint(api)

# Browser preflight (automatic, before your JSON POST):
OPTIONS /api/v1/users
Origin: http://localhost:3000
Access-Control-Request-Method: POST
Access-Control-Request-Headers: content-type

# flask-cors replies:
Access-Control-Allow-Origin: http://localhost:3000
Access-Control-Allow-Methods: GET, POST, PATCH, DELETE
Access-Control-Allow-Headers: content-type
Access-Control-Allow-Credentials: true

# Only THEN does the browser send the real POST.

# pip install "flask[async]" httpx
import asyncio, httpx
from flask import Flask

app = Flask(__name__)

# ── The legit use case: parallel outbound calls in ONE request ──
@app.get("/dashboard-data")
async def dashboard_data():
    async with httpx.AsyncClient(timeout=5) as client:
        users_task  = client.get("https://api.internal/users/stats")
        sales_task  = client.get("https://api.internal/sales/stats")
        alerts_task = client.get("https://api.internal/alerts")

        users, sales, alerts = await asyncio.gather(
            users_task, sales_task, alerts_task
        )                       # 3 calls in the time of the slowest one

    return {
        "users":  users.json(),
        "sales":  sales.json(),
        "alerts": alerts.json(),
    }

# Sync version would take t1 + t2 + t3.
# Async version takes max(t1, t2, t3).
# But NOTE: the worker is still busy for this whole request —
# other users' requests wait for a free worker either way.

# How a request flows in production:
#
# Client → nginx (TLS, static files, rate limit)
#            │
#            └→ gunicorn (WSGI process manager)
#                  ├── worker 1 ─┐
#                  ├── worker 2 ─┼── your Flask app (create_app())
#                  ├── worker 3 ─┤
#                  └── worker 4 ─┘
#                        │
#                  PostgreSQL / Redis / Celery workers
#
# Run command:
# gunicorn "app:create_app()" -w 4 -b 0.0.0.0:8000

my-flask-app/
│
├── app/                        # all application code (the package)
│   ├── __init__.py             # create_app() factory lives here
│   ├── config.py               # Dev/Test/Prod config classes
│   ├── extensions.py           # db, migrate, login_manager (unbound)
│   │
│   ├── models/                 # SQLAlchemy models (DB tables)
│   │   ├── __init__.py
│   │   ├── user.py
│   │   └── post.py
│   │
│   ├── blueprints/             # route handlers, grouped by feature
│   │   ├── __init__.py
│   │   ├── main.py             # /, /about — public pages
│   │   ├── auth.py             # /auth/login, /auth/register
│   │   └── api.py              # /api/v1/* — JSON endpoints
│   │
│   ├── services/               # business logic (no HTTP in here)
│   │   ├── __init__.py
│   │   └── user_service.py
│   │
│   ├── templates/              # Jinja2 templates
│   │   ├── base.html
│   │   ├── auth/login.html
│   │   └── errors/404.html
│   │
│   └── static/                 # css / js / images
│       └── css/style.css
│
├── migrations/                 # Flask-Migrate / Alembic scripts
├── tests/                      # mirrors app/ structure
│   ├── conftest.py             # fixtures: app, client, db
│   ├── test_auth.py
│   └── test_api.py
│
├── wsgi.py                     # entry point: app = create_app()
├── .env                        # secrets (NEVER commit)
├── .env.example                # template for .env (commit this)
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── .gitignore

# One class per environment. Secrets always come from env vars.
import os
from dotenv import load_dotenv

load_dotenv()                                   # reads .env in development

class BaseConfig:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-only-change-me")
    SQLALCHEMY_TRACK_MODIFICATIONS = False

class DevConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL", "sqlite:///dev.db")

class TestConfig(BaseConfig):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
    WTF_CSRF_ENABLED = False                    # simpler form tests

class ProdConfig(BaseConfig):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ["DATABASE_URL"]   # required!
    SESSION_COOKIE_SECURE = True
    REMEMBER_COOKIE_SECURE = True

config_map = {"dev": DevConfig, "test": TestConfig, "prod": ProdConfig}

# Instantiate extensions WITHOUT an app.
# Everyone imports from here → zero circular imports.
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from flask_login import LoginManager

db            = SQLAlchemy()
migrate       = Migrate()
login_manager = LoginManager()
login_manager.login_view = "auth.login"        # blueprint.endpoint

import os
from flask import Flask, render_template
from .config import config_map
from .extensions import db, migrate, login_manager

def create_app(config_name: str | None = None) -> Flask:
    config_name = config_name or os.environ.get("FLASK_CONFIG", "dev")

    app = Flask(__name__)
    app.config.from_object(config_map[config_name])

    # ── 1. Bind extensions ──
    db.init_app(app)
    migrate.init_app(app, db)
    login_manager.init_app(app)

    # ── 2. Register blueprints (import here → no cycles) ──
    from .blueprints.main import bp as main_bp
    from .blueprints.auth import bp as auth_bp
    from .blueprints.api  import bp as api_bp
    app.register_blueprint(main_bp)
    app.register_blueprint(auth_bp, url_prefix="/auth")
    app.register_blueprint(api_bp,  url_prefix="/api/v1")

    # ── 3. Error pages ──
    @app.errorhandler(404)
    def not_found(e):
        return render_template("errors/404.html"), 404

    # ── 4. Shell context (flask shell gets db + models for free) ──
    from .models.user import User
    @app.shell_context_processor
    def shell_ctx():
        return {"db": db, "User": User}

    return app

from flask_login import UserMixin
from werkzeug.security import generate_password_hash, check_password_hash
from ..extensions import db, login_manager

class User(UserMixin, db.Model):
    __tablename__ = "users"
    id       = db.Column(db.Integer, primary_key=True)
    email    = db.Column(db.String(255), unique=True, index=True, nullable=False)
    pw_hash  = db.Column(db.String(255), nullable=False)
    role     = db.Column(db.String(20), default="user")

    # Never store the plain password — property blocks reads too
    @property
    def password(self):
        raise AttributeError("password is write-only")

    @password.setter
    def password(self, plain):
        self.pw_hash = generate_password_hash(plain)

    def check_password(self, plain) -> bool:
        return check_password_hash(self.pw_hash, plain)

@login_manager.user_loader
def load_user(user_id):
    return db.session.get(User, int(user_id))

# Pure business logic — no request/response objects.
# Reusable from routes, CLI commands, Celery tasks, and tests.
from ..extensions import db
from ..models.user import User

class EmailTaken(Exception):
    pass

def register_user(email: str, password: str) -> User:
    exists = db.session.execute(
        db.select(User).filter_by(email=email)
    ).scalar_one_or_none()
    if exists:
        raise EmailTaken(email)

    user = User(email=email, password=password)   # setter hashes it
    db.session.add(user)
    db.session.commit()
    return user

def authenticate(email: str, password: str) -> User | None:
    user = db.session.execute(
        db.select(User).filter_by(email=email)
    ).scalar_one_or_none()
    if user and user.check_password(password):
        return user
    return None

# Routes stay THIN: parse → call service → respond.
from flask import Blueprint, render_template, request, redirect, url_for, flash
from flask_login import login_user, logout_user, login_required
from ..services import user_service

bp = Blueprint("auth", __name__)

@bp.route("/register", methods=["GET", "POST"])
def register():
    if request.method == "POST":
        try:
            user_service.register_user(request.form["email"],
                                       request.form["password"])
            flash("Account created — log in!")
            return redirect(url_for("auth.login"))
        except user_service.EmailTaken:
            flash("Email already registered")
    return render_template("auth/register.html")

@bp.route("/login", methods=["GET", "POST"])
def login():
    if request.method == "POST":
        user = user_service.authenticate(request.form["email"],
                                         request.form["password"])
        if user:
            login_user(user)
            return redirect(url_for("main.index"))
        flash("Invalid credentials")
    return render_template("auth/login.html")

@bp.post("/logout")
@login_required
def logout():
    logout_user()
    return redirect(url_for("auth.login"))

# wsgi.py — what servers import
from app import create_app
app = create_app()

# Run dev:   flask --app wsgi run --debug
# Run prod:  gunicorn wsgi:app -w 4 -b 0.0.0.0:8000


# ── .env (NEVER commit — real secrets) ──
# SECRET_KEY=long-random-string-here
# DATABASE_URL=postgresql://myuser:mypass@localhost:5432/mydb
# FLASK_CONFIG=dev

# ── .env.example (commit this template) ──
# SECRET_KEY=change-me
# DATABASE_URL=postgresql://user:password@localhost:5432/dbname
# FLASK_CONFIG=dev

# POST /auth/register  (email=y@dev.com, password=secret123)
#
# 1. gunicorn worker hands the WSGI request to Flask
# 2. Werkzeug routing → blueprint "auth", view register()
# 3. before_request hooks run (auth gates, timers…)
# 4. View parses request.form → calls user_service.register_user()
# 5. Service enforces business rules (email unique),
#    hashes password, db.session.add + commit
# 6. View redirects → after_request hooks add headers
# 7. teardown removes the DB session
#
# Request:  Client → nginx → gunicorn → routing → blueprint → service → db
# Response: db → service → blueprint → hooks → gunicorn → nginx → Client

# pip install pytest
import pytest
from app import create_app
from app.extensions import db as _db

@pytest.fixture
def app():
    """Fresh app + clean in-memory DB for every test."""
    app = create_app("test")
    with app.app_context():
        _db.create_all()
        yield app
        _db.session.remove()
        _db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

@pytest.fixture
def user(app):
    """A ready-made user for auth tests."""
    from app.models.user import User
    u = User(email="y@dev.com", password="secret123")
    _db.session.add(u)
    _db.session.commit()
    return u

@pytest.fixture
def logged_in(client, user):
    client.post("/auth/login",
                data={"email": "y@dev.com", "password": "secret123"})
    return client

def test_health(client):
    resp = client.get("/api/v1/health")
    assert resp.status_code == 200
    assert resp.get_json() == {"status": "ok"}

def test_create_user(client):
    resp = client.post("/api/v1/users",
                       json={"name": "Yatin", "email": "new@dev.com"})
    assert resp.status_code == 201
    body = resp.get_json()
    assert body["name"] == "Yatin"
    assert "id" in body

def test_validation_error(client):
    resp = client.post("/api/v1/users", json={"name": ""})
    assert resp.status_code == 400
    assert "error" in resp.get_json()

def test_missing_user_404(client):
    assert client.get("/api/v1/users/9999").status_code == 404

def test_protected_requires_login(client):
    resp = client.get("/dashboard")
    assert resp.status_code == 302                 # bounced to login
    assert "/auth/login" in resp.headers["Location"]

def test_dashboard_when_logged_in(logged_in):
    resp = logged_in.get("/dashboard")
    assert resp.status_code == 200
    assert b"y@dev.com" in resp.data

def test_login_flow_end_to_end(client, user):
    resp = client.post("/auth/login",
                       data={"email": "y@dev.com", "password": "secret123"},
                       follow_redirects=True)
    assert resp.status_code == 200

# ── Unit-testing a helper that needs the request context ──
def test_helper_with_request_context(app):
    from flask import request
    with app.test_request_context("/search?q=flask"):
        assert request.args["q"] == "flask"

# ── Pre-seeding the session directly ──
def test_session_write(client):
    with client.session_transaction() as sess:
        sess["user_id"] = 42
    resp = client.get("/whoami")
    assert resp.get_json()["user_id"] == 42

pytest -v                      # run everything
pytest tests/test_api.py -k login    # just login tests
pip install pytest-cov
pytest --cov=app --cov-report=term-missing   # coverage report

FROM python:3.12-slim

ENV PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

WORKDIR /app

# Install deps first → Docker layer caching
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# Non-root user (security best practice)
RUN adduser --disabled-password --no-create-home appuser
USER appuser

EXPOSE 8000
CMD ["gunicorn", "wsgi:app", \
     "-w", "4", \
     "--threads", "2", \
     "-b", "0.0.0.0:8000", \
     "--access-logfile", "-"]

# docker compose up --build  →  app + Postgres + Redis
services:
  web:
    build: .
    ports:
      - "8000:8000"
    env_file: .env
    depends_on: [db, redis]
    command: >
      sh -c "flask db upgrade &&
             gunicorn wsgi:app -w 4 -b 0.0.0.0:8000"

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: myuser
      POSTGRES_PASSWORD: mypassword
      POSTGRES_DB: mydb
    volumes:
      - pgdata:/var/lib/postgresql/data

  redis:
    image: redis:7-alpine

volumes:
  pgdata:

server {
    listen 80;
    server_name myapp.com;

    location /static/ {                # nginx serves assets directly
        alias /srv/myapp/app/static/;
        expires 30d;
    }

    location / {                       # everything else → gunicorn
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host              $host;
        proxy_set_header X-Real-IP         $remote_addr;
        proxy_set_header X-Forwarded-For   $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }
}
# Then in Flask:  app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1)

# pip install flask-caching redis
from flask import Flask
from flask_caching import Cache

app = Flask(__name__)
app.config.update(
    CACHE_TYPE="RedisCache",
    CACHE_REDIS_URL="redis://localhost:6379/0",
    CACHE_DEFAULT_TIMEOUT=300,
)
cache = Cache(app)                 # factory apps: cache.init_app(app)

# ── Cache an entire view for 5 minutes ──
@app.get("/stats")
@cache.cached(timeout=300)
def stats():
    return {"totals": run_expensive_report()}     # hits DB 1×/5min

# ── Cache per-argument (memoize) ──
@cache.memoize(timeout=600)
def get_user_profile(user_id):
    return db.session.get(User, user_id)          # cached per user_id

@app.post("/users/<int:user_id>")
def update_user(user_id):
    ...  # write to DB
    cache.delete_memoized(get_user_profile, user_id)   # invalidate!
    return {"ok": True}

# pip install redis flask-limiter
import json, redis
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address

r = redis.from_url("redis://localhost:6379/0", decode_responses=True)

# ── Cache-aside pattern: check → miss → DB → store ──
@app.get("/products/<int:pid>")
def product(pid):
    key = f"product:{pid}"
    cached = r.get(key)
    if cached:
        return json.loads(cached)                 # cache HIT — no DB

    prod = fetch_product_from_db(pid)             # cache MISS
    if prod is None:
        return {"error": "not found"}, 404
    r.set(key, json.dumps(prod), ex=300)          # store with TTL
    return prod

# Invalidate on write:
#   r.delete(f"product:{pid}")

# ── Rate limiting (shared across ALL workers) ──
limiter = Limiter(
    get_remote_address,                           # limit per client IP
    app=app,
    storage_uri="redis://localhost:6379/1",
    default_limits=["200 per hour"],
)

@app.get("/api/search")
@limiter.limit("10 per minute")                   # stricter on this route
def search():
    return {"results": []}
# Over the limit → automatic 429 Too Many Requests

# pip install celery redis
# app/celery_utils.py — the official Flask docs pattern
from celery import Celery, Task

def celery_init_app(app):
    class FlaskTask(Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():            # DB + config work in tasks
                return self.run(*args, **kwargs)

    celery_app = Celery(app.name, task_cls=FlaskTask)
    celery_app.config_from_object(app.config["CELERY"])
    celery_app.set_default()
    app.extensions["celery"] = celery_app
    return celery_app


# In create_app():
app.config["CELERY"] = dict(
    broker_url="redis://localhost:6379/0",
    result_backend="redis://localhost:6379/1",
    task_ignore_result=False,
)
celery_init_app(app)


# make_celery.py — what the worker command imports
from app import create_app
flask_app = create_app()
celery_app = flask_app.extensions["celery"]

# app/tasks.py
from celery import shared_task

@shared_task(ignore_result=True)
def send_welcome_email(user_id):
    user = db.session.get(User, user_id)       # re-fetch by id
    smtp_send(to=user.email, subject="Welcome!")

@shared_task(bind=True, max_retries=3)
def call_flaky_api(self, order_id):
    try:
        return post_to_partner(order_id)
    except TimeoutError as exc:
        # retry in 30s, then 60s, then 120s
        raise self.retry(exc=exc, countdown=30 * (2 ** self.request.retries))

@shared_task
def generate_report(month):
    data = crunch_numbers(month)               # takes minutes — who cares!
    return {"rows": len(data)}


# app/blueprints/api.py — fire and track
from celery.result import AsyncResult
from ..tasks import send_welcome_email, generate_report

@bp.post("/signup")
def signup():
    user = user_service.register_user(...)
    send_welcome_email.delay(user.id)          # returns in microseconds
    return {"id": user.id}, 201                # response is instant

@bp.post("/reports")
def start_report():
    result = generate_report.delay("2026-08")
    return {"task_id": result.id}, 202          # 202 Accepted

@bp.get("/reports/<task_id>")
def report_status(task_id):
    res = AsyncResult(task_id)
    return {
        "state": res.state,                     # PENDING / STARTED / SUCCESS / FAILURE
        "result": res.result if res.successful() else None,
    }

# Terminal 1 — Redis (broker + results)
docker run -p 6379:6379 redis:7-alpine

# Terminal 2 — the web app
gunicorn wsgi:app -w 4

# Terminal 3 — celery worker(s)
celery -A make_celery worker --loglevel INFO --concurrency 4

# Optional — scheduled tasks (Celery beat)
celery -A make_celery beat --loglevel INFO

# Optional — web dashboard for tasks
pip install flower
celery -A make_celery flower          # → http://localhost:5555