import sqlite3

conn = sqlite3.connect('app.db')
cursor = conn.cursor()

# 手动拼接 SQL——危险！
cursor.execute("SELECT * FROM users WHERE username = '%s'" % username)

# 即使使用参数化查询，代码依然繁琐
cursor.execute(
    "INSERT INTO todo (title, done, user_id) VALUES (?, ?, ?)",
    (title, done, user_id)
)
conn.commit()

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timezone

# 创建 Flask 应用
app = Flask(__name__)

# 🔑 配置数据库连接
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'  # 数据库位置
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False         # 关闭事件跟踪（节省内存）

# 创建数据库实例
db = SQLAlchemy(app)

class User(db.Model):
    """用户模型——对应数据库中的 user 表"""
    __tablename__ = 'user'  # 表名（不指定则默认类名小写）
    
    # 定义列（字段）
    id = db.Column(db.Integer, primary_key=True)       # 主键，自增
    username = db.Column(db.String(80), unique=True, nullable=False)  # 唯一，非空
    email = db.Column(db.String(120), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    
    # 关系（一对多）
    todos = db.relationship('Todo', backref='user', lazy=True)
    
    def __repr__(self):
        return f'<User {self.username}>'

class Todo(db.Model):
    __tablename__ = 'todo'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    done = db.Column(db.Boolean, default=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    # user_id 是外键，指向 user 表的 id
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'done': self.done,
            'user_id': self.user_id,
            'created_at': self.created_at.isoformat() if self.created_at else None
        }

# 正向访问：通过用户获取他的所有 todo
user = User.query.get(1)
user.todos  # ➜ [Todo, Todo, ...]

# 反向访问：通过 todo 获取它的所有者
todo = Todo.query.get(1)
todo.user   # ➜ <User xxx>
# ^ 这就是 backref='user' 的作用——给 Todo 类添加了 .user 属性

# ──────────────── 创建 ────────────────

user = User(username='alice', email='alice@example.com')
db.session.add(user)      # 添加到会话
db.session.commit()       # 提交事务（写入数据库）

# 批量创建
todo1 = Todo(title='学 Flask', user=user)
todo2 = Todo(title='学 SQLAlchemy', user=user)
db.session.add_all([todo1, todo2])
db.session.commit()

# ──────────────── 读取 ────────────────

# 获取全部
User.query.all()

# 按主键获取
user = User.query.get(1)

# 按条件过滤
users = User.query.filter_by(username='alice').all()      # 精确匹配
users = User.query.filter(User.email.endswith('@gmail.com')).all()  # 模糊匹配
user = User.query.filter_by(username='alice').first()     # 取第一个

# 排序
todos = Todo.query.order_by(Todo.created_at.desc()).all()  # 倒序

# 分页
todos = Todo.query.paginate(page=1, per_page=10, error_out=False)

# 复杂查询
results = db.session.execute(
    db.select(User).where(User.username.in_(['alice', 'bob']))
).scalars().all()

# ──────────────── 更新 ────────────────

user = User.query.get(1)
user.email = 'newemail@example.com'  # 直接修改属性
db.session.commit()                   # 提交更新

# ──────────────── 删除 ────────────────

user = User.query.get(1)
db.session.delete(user)
db.session.commit()

pip install Flask-Migrate

from flask_migrate import Migrate

migrate = Migrate(app, db)

# 终端操作
# flask db init        # 初始化迁移仓库
# flask db migrate     # 自动检测模型变化，生成迁移脚本
# flask db upgrade     # 应用迁移到数据库
# flask db downgrade   # 回滚迁移

from werkzeug.security import generate_password_hash, check_password_hash

# 注册：存储哈希值
password_hash = generate_password_hash('my_password123')
# ➜ 'pbkdf2:sha256:600000$...'  （自动加盐，默认 600000 次迭代）

# 登录：验证密码
is_valid = check_password_hash(password_hash, 'my_password123')  # ➜ True
is_valid = check_password_hash(password_hash, 'wrong_password')  # ➜ False

from flask_login import LoginManager, UserMixin, login_user, logout_user, login_required, current_user

# ──────────────── 初始化 ────────────────
login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'  # 未登录时重定向到登录页面

# ──────────────── 用户加载回调 ────────────────
@login_manager.user_loader
def load_user(user_id):
    """Flask-Login 需要这个函数来从 session 中恢复用户"""
    return User.query.get(int(user_id))

# ──────────────── 用户模型 ────────────────
class User(UserMixin, db.Model):
    """UserMixin 提供了 Flask-Login 需要的 is_authenticated/is_active/is_anonymous/get_id 方法"""
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    password_hash = db.Column(db.String(256), nullable=False)

# ──────────────── 登录路由 ────────────────
@app.route('/login', methods=['POST'])
def login():
    data = request.get_json()
    user = User.query.filter_by(username=data['username']).first()
    
    if user and check_password_hash(user.password_hash, data['password']):
        login_user(user)               # 🔑 在 session 中标记用户已登录
        return jsonify({'message': '登录成功'})
    return jsonify({'error': '用户名或密码错误'}), 401

# ──────────────── 登出路由 ────────────────
@app.route('/logout', methods=['POST'])
@login_required  # 需要登录才能访问
def logout():
    logout_user()  # 🔓 清除 session 中的登录标记
    return jsonify({'message': '已登出'})

# ──────────────── 保护路由 ────────────────
@app.route('/profile')
@login_required  # 未登录用户会被重定向到 login_manager.login_view
def profile():
    return jsonify({
        'username': current_user.username  # current_user 是当前登录用户
    })

import jwt
from datetime import datetime, timedelta, timezone

# ──────────────── 生成 Token ────────────────
def create_token(user_id):
    payload = {
        'user_id': user_id,
        'exp': datetime.now(timezone.utc) + timedelta(hours=24),  # 过期时间
        'iat': datetime.now(timezone.utc)                         # 签发时间
    }
    return jwt.encode(payload, app.config['SECRET_KEY'], algorithm='HS256')

# ──────────────── 验证 Token（装饰器） ────────────────
from functools import wraps

def token_required(f):
    @wraps(f)
    def decorated(*args, **kwargs):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')
        if not token:
            return jsonify({'error': '缺少 Token'}), 401
        try:
            payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
            current_user = User.query.get(payload['user_id'])
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token 已过期'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': '无效的 Token'}), 401
        return f(current_user, *args, **kwargs)
    return decorated

# ──────────────── 使用 ────────────────
@app.route('/api/todos', methods=['GET'])
@token_required
def get_todos(current_user):
    todos = Todo.query.filter_by(user_id=current_user.id).all()
    return jsonify([t.to_dict() for t in todos])

# ✅ 成功响应
{
    "data": [...],           # 主体数据
    "meta": {                # 元信息（分页等）
        "page": 1,
        "per_page": 10,
        "total": 42
    },
    "message": "获取成功"    # 可选的提示信息
}

# ❌ 错误响应
{
    "error": "资源不存在",             # 人类可读的错误描述
    "code": "NOT_FOUND",              # 机器可读的错误码
    "details": {"id": "用户 ID 无效"}  # 详细的字段级错误
}

from flask import jsonify

# 200 — 获取成功
return jsonify({'data': todos}), 200

# 201 — 创建成功
return jsonify({'data': todo.to_dict(), 'message': '创建成功'}), 201

# 204 — 删除成功（无内容）
return '', 204

# 400 — 请求参数错误
return jsonify({'error': 'title 字段是必填的'}), 400

# 404 — 资源不存在
return jsonify({'error': '任务不存在'}), 404

# 409 — 资源冲突（如用户名已存在）
return jsonify({'error': '用户名已被注册'}), 409

#!/usr/bin/env python3
"""
TODO API 完整应用
集成了 SQLAlchemy 数据库 + JWT 认证 + RESTful 接口
"""

from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from datetime import datetime, timedelta, timezone
from functools import wraps
import jwt

# ──────────────── 应用初始化 ────────────────

app = Flask(__name__)

app.config['SECRET_KEY'] = 'your-secret-key-change-in-production'
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///todo.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)


# ──────────────── 数据库模型 ────────────────

class User(db.Model):
    __tablename__ = 'user'
    
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(256), nullable=False)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    
    # 关系
    todos = db.relationship('Todo', backref='user', lazy='dynamic')
    
    def to_dict(self):
        return {
            'id': self.id,
            'username': self.username,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'todo_count': self.todos.count()
        }


class Todo(db.Model):
    __tablename__ = 'todo'
    
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(200), nullable=False)
    description = db.Column(db.Text, default='')
    done = db.Column(db.Boolean, default=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False, index=True)
    created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
    updated_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc),
                           onupdate=lambda: datetime.now(timezone.utc))
    
    def to_dict(self):
        return {
            'id': self.id,
            'title': self.title,
            'description': self.description,
            'done': self.done,
            'user_id': self.user_id,
            'created_at': self.created_at.isoformat() if self.created_at else None,
            'updated_at': self.updated_at.isoformat() if self.updated_at else None,
        }


# ──────────────── JWT 认证 ────────────────

from werkzeug.security import generate_password_hash, check_password_hash


def create_token(user_id):
    """生成 JWT Token"""
    payload = {
        'user_id': user_id,
        'exp': datetime.now(timezone.utc) + timedelta(hours=24),
        'iat': datetime.now(timezone.utc),
    }
    return jwt.encode(payload, app.config['SECRET_KEY'], algorithm='HS256')


def token_required(f):
    """JWT 验证装饰器"""
    @wraps(f)
    def decorated(*args, **kwargs):
        auth_header = request.headers.get('Authorization', '')
        
        if not auth_header.startswith('Bearer '):
            return jsonify({'error': '缺少或无效的 Authorization 头'}), 401
        
        token = auth_header.split(' ')[1]
        
        try:
            payload = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
            current_user = User.query.get(payload['user_id'])
            if not current_user:
                return jsonify({'error': '用户不存在'}), 401
        except jwt.ExpiredSignatureError:
            return jsonify({'error': 'Token 已过期，请重新登录'}), 401
        except jwt.InvalidTokenError:
            return jsonify({'error': '无效的 Token'}), 401
        
        return f(current_user, *args, **kwargs)
    return decorated


# ──────────────── 认证路由 ────────────────

@app.route('/api/auth/register', methods=['POST'])
def register():
    """用户注册"""
    data = request.get_json()
    
    # 参数验证
    if not data or not data.get('username') or not data.get('password'):
        return jsonify({'error': '用户名和密码是必填的'}), 400
    
    username = data['username'].strip()
    password = data['password']
    
    if len(username) < 3 or len(username) > 20:
        return jsonify({'error': '用户名长度必须在 3-20 个字符之间'}), 400
    
    if len(password) < 6:
        return jsonify({'error': '密码长度不能少于 6 个字符'}), 400
    
    # 检查用户名是否已存在
    if User.query.filter_by(username=username).first():
        return jsonify({'error': '用户名已被注册'}), 409
    
    # 创建用户
    user = User(
        username=username,
        password_hash=generate_password_hash(password)
    )
    db.session.add(user)
    db.session.commit()
    
    # 生成 Token
    token = create_token(user.id)
    
    return jsonify({
        'message': '注册成功',
        'data': {
            'user': user.to_dict(),
            'token': token
        }
    }), 201


@app.route('/api/auth/login', methods=['POST'])
def login():
    """用户登录"""
    data = request.get_json()
    
    if not data or not data.get('username') or not data.get('password'):
        return jsonify({'error': '用户名和密码是必填的'}), 400
    
    user = User.query.filter_by(username=data['username'].strip()).first()
    
    if not user or not check_password_hash(user.password_hash, data['password']):
        return jsonify({'error': '用户名或密码错误'}), 401
    
    token = create_token(user.id)
    
    return jsonify({
        'message': '登录成功',
        'data': {
            'user': user.to_dict(),
            'token': token
        }
    })


@app.route('/api/auth/profile', methods=['GET'])
@token_required
def get_profile(current_user):
    """获取当前用户信息"""
    return jsonify({'data': current_user.to_dict()})


# ──────────────── TODO CRUD 路由 ────────────────

@app.route('/api/todos', methods=['GET'])
@token_required
def get_todos(current_user):
    """获取当前用户的所有 TODO"""
    # 支持过滤
    status = request.args.get('status')  # 'done' 或 'active'
    page = request.args.get('page', 1, type=int)
    per_page = request.args.get('per_page', 10, type=int)
    
    query = Todo.query.filter_by(user_id=current_user.id)
    
    if status == 'done':
        query = query.filter_by(done=True)
    elif status == 'active':
        query = query.filter_by(done=False)
    
    # 分页
    pagination = query.order_by(Todo.created_at.desc()).paginate(
        page=page, per_page=per_page, error_out=False
    )
    
    return jsonify({
        'data': [todo.to_dict() for todo in pagination.items],
        'meta': {
            'page': pagination.page,
            'per_page': pagination.per_page,
            'total': pagination.total,
            'pages': pagination.pages
        }
    })


@app.route('/api/todos/<int:todo_id>', methods=['GET'])
@token_required
def get_todo(current_user, todo_id):
    """获取单个 TODO"""
    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
    if not todo:
        return jsonify({'error': '任务不存在'}), 404
    return jsonify({'data': todo.to_dict()})


@app.route('/api/todos', methods=['POST'])
@token_required
def create_todo(current_user):
    """创建 TODO"""
    data = request.get_json()
    
    if not data or not data.get('title', '').strip():
        return jsonify({'error': '标题是必填的'}), 400
    
    title = data['title'].strip()
    if len(title) > 200:
        return jsonify({'error': '标题长度不能超过 200 个字符'}), 400
    
    todo = Todo(
        title=title,
        description=data.get('description', ''),
        user_id=current_user.id
    )
    db.session.add(todo)
    db.session.commit()
    
    return jsonify({
        'message': '创建成功',
        'data': todo.to_dict()
    }), 201


@app.route('/api/todos/<int:todo_id>', methods=['PUT'])
@token_required
def update_todo(current_user, todo_id):
    """更新 TODO（全量更新）"""
    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
    if not todo:
        return jsonify({'error': '任务不存在'}), 404
    
    data = request.get_json()
    if not data:
        return jsonify({'error': '请求体为空'}), 400
    
    todo.title = data.get('title', todo.title).strip()
    todo.description = data.get('description', todo.description)
    todo.done = data.get('done', todo.done)
    
    db.session.commit()
    
    return jsonify({
        'message': '更新成功',
        'data': todo.to_dict()
    })


@app.route('/api/todos/<int:todo_id>', methods=['PATCH'])
@token_required
def patch_todo(current_user, todo_id):
    """部分更新 TODO"""
    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
    if not todo:
        return jsonify({'error': '任务不存在'}), 404
    
    data = request.get_json()
    if not data:
        return jsonify({'error': '请求体为空'}), 400
    
    # 只更新提供的字段
    if 'title' in data:
        todo.title = data['title'].strip()
    if 'description' in data:
        todo.description = data['description']
    if 'done' in data:
        todo.done = data['done']
    
    db.session.commit()
    
    return jsonify({
        'message': '更新成功',
        'data': todo.to_dict()
    })


@app.route('/api/todos/<int:todo_id>', methods=['DELETE'])
@token_required
def delete_todo(current_user, todo_id):
    """删除 TODO"""
    todo = Todo.query.filter_by(id=todo_id, user_id=current_user.id).first()
    if not todo:
        return jsonify({'error': '任务不存在'}), 404
    
    db.session.delete(todo)
    db.session.commit()
    
    return '', 204


# ──────────────── 错误处理 ────────────────

@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': '接口不存在'}), 404


@app.errorhandler(405)
def method_not_allowed(error):
    return jsonify({'error': '不支持该请求方法'}), 405


@app.errorhandler(500)
def internal_error(error):
    db.session.rollback()  # 回滚事务
    return jsonify({'error': '服务器内部错误'}), 500


# ──────────────── 启动 ────────────────

if __name__ == '__main__':
    with app.app_context():
        db.create_all()  # 创建所有表
    app.run(debug=True, host='0.0.0.0', port=5000)

# 1. 注册用户
curl -X POST http://localhost:5000/api/auth/register \
  -H "Content-Type: application/json" \
  -d '{"username":"alice","password":"secret123"}'

# 2. 登录（获取 Token）
curl -X POST http://localhost:5000/api/auth/login \
  -H "Content-Type: application/json" \
  -d '{"username":"alice","password":"secret123"}'
# 保存返回的 token

# 3. 创建 TODO
TOKEN="<上面的 token>"
curl -X POST http://localhost:5000/api/todos \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"title":"学习 Flask 进阶","description":"完成 Day 066 所有内容"}'

# 4. 获取 TODO 列表
curl http://localhost:5000/api/todos \
  -H "Authorization: Bearer $TOKEN"

# 5. 更新 TODO
curl -X PATCH http://localhost:5000/api/todos/1 \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer $TOKEN" \
  -d '{"done":true}'

# 6. 删除 TODO
curl -X DELETE http://localhost:5000/api/todos/1 \
  -H "Authorization: Bearer $TOKEN"