Plain Text
flask_demo/
├── app.py # 主程序
├── requirements.txt # 依赖
└── .env # 数据库配置

txt
flask
flask-sqlalchemy
pymysql
python-dotenv
cryptography

env
DB_HOST=localhost
DB_USER=root
DB_PASSWORD=你的密码
DB_NAME=flask_demo
DB_PORT=3306

python
from flask import Flask, request, jsonify
from flask_sqlalchemy import SQLAlchemy
from dotenv import load_dotenv
import os

# 加载环境变量
load_dotenv()

# 初始化 Flask
app = Flask(__name__)

# 连接 MySQL
app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

# 初始化数据库
db = SQLAlchemy(app)

# ==================== 数据库模型 ====================
class User(db.Model):
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(50))
email = db.Column(db.String(100), unique=True)
age = db.Column(db.Integer)

# ==================== 接口（CRUD） ====================

# 1. 创建用户
@app.route('/user', methods=['POST'])
def create_user():
data = request.json
user = User(name=data['name'], email=data['email'], age=data.get('age'))
db.session.add(user)
db.session.commit()
return jsonify({"msg": "创建成功", "id": user.id})

# 2. 查询所有用户
@app.route('/users', methods=['GET'])
def get_users():
users = User.query.all()
res = [{"id": u.id, "name": u.name, "email": u.email, "age": u.age} for u in users]
return jsonify(res)

# 3. 查询单个用户
@app.route('/user/<int:id>', methods=['GET'])
def get_user(id):
user = User.query.get(id)
if not user:
return jsonify({"msg": "用户不存在"}), 404
return jsonify({"id": user.id, "name": user.name, "email": user.email, "age": user.age})

# 4. 更新用户
@app.route('/user/<int:id>', methods=['PUT'])
def update_user(id):
user = User.query.get(id)
if not user:
return jsonify({"msg": "用户不存在"}), 404
data = request.json
user.name = data.get('name', user.name)
user.email = data.get('email', user.email)
user.age = data.get('age', user.age)
db.session.commit()
return jsonify({"msg": "更新成功"})

# 5. 删除用户
@app.route('/user/<int:id>', methods=['DELETE'])
def delete_user(id):
user = User.query.get(id)
if not user:
return jsonify({"msg": "用户不存在"}), 404
db.session.delete(user)
db.session.commit()
return jsonify({"msg": "删除成功"})

# 启动时创建表
with app.app_context():
db.create_all()

# 启动
if __name__ == '__main__':
app.run(debug=True)

bash
pip install -r requirements.txt

bash
python app.py

Plain Text
http://127.0.0.1:5000