API 管理双十一促销活动

import redis
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def get_product_info(product_id):
    cache_key = f"product:{product_id}"
    product_info = r.get(cache_key)
    if product_info is None:
        # 模拟从数据库获取数据
        product_info = fetch_from_db(product_id)
        r.setex(cache_key, 3600, product_info)  # 缓存1小时
    return product_info

def fetch_from_db(product_id):
    # 模拟数据库查询
    time.sleep(0.1)
    return f"Info of product {product_id}"

import sqlite3

def update_product_stock(product_id, quantity):
    conn = sqlite3.connect('products.db')
    cursor = conn.cursor()
    cursor.execute("SELECT stock FROM products WHERE id = ? FOR UPDATE", (product_id,))
    current_stock = cursor.fetchone()[0]
    new_stock = current_stock - quantity
    cursor.execute("UPDATE products SET stock = ? WHERE id = ?", (new_stock, product_id))
    conn.commit()
    conn.close()

import jwt
from flask import request, jsonify

SECRET_KEY = 'your_secret_key'

def authenticate(func):
    def wrapper(*args, **kwargs):
        token = request.headers.get('Authorization')
        if not token:
            return jsonify({"error": "Unauthorized"}), 401
        try:
            data = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        except jwt.ExpiredSignatureError:
            return jsonify({"error": "Token has expired"}), 401
        except jwt.InvalidTokenError:
            return jsonify({"error": "Invalid token"}), 401
        return func(*args, **kwargs)
    return wrapper

@app.route('/protected')
@authenticate
def protected():
    return jsonify({"message": "This is a protected endpoint"})