Python 连接 MySQL 的「实战级开发教程」

一、准备工作必须1️⃣ 安装 MySQLWindowshttps://dev.mysql.com/downloads/installer/macOSbrew install mysqlLinuxsudo apt install mysql-server确保你能登录mysql -u root -p2️⃣ 创建数据库和表示例CREATE DATABASE flask_demo DEFAULT CHARSET utf8mb4; USE flask_demo; CREATE TABLE users ( id INT PRIMARY KEY AUTO_INCREMENT, username VARCHAR(50) NOT NULL UNIQUE, password VARCHAR(100) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); INSERT INTO users (username, password) VALUES (admin, 123456);3️⃣ 安装 Python MySQL 驱动重点✅强烈推荐 PyMySQL最常用pip install pymysql如果你之前 pip 有问题用python -m pip install pymysql二、Python 连接 MySQL最基础版import pymysql conn pymysql.connect( hostlocalhost, port3306, userroot, password你的密码, databaseflask_demo, charsetutf8mb4 ) cursor conn.cursor() cursor.execute(SELECT * FROM users) rows cursor.fetchall() for row in rows: print(row) cursor.close() conn.close()✅ 成功打印用户信息说明连接 OK三、封装一个 DB 工具类推荐做法 ✅import pymysql class DB: def __init__(self): self.conn pymysql.connect( hostlocalhost, port3306, userroot, password你的密码, databaseflask_demo, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor # 返回字典 ) def query(self, sql, paramsNone): with self.conn.cursor() as cursor: cursor.execute(sql, params or ()) return cursor.fetchall() def execute(self, sql, paramsNone): with self.conn.cursor() as cursor: cursor.execute(sql, params or ()) self.conn.commit() return cursor.lastrowid def close(self): self.conn.close()四、查询示例登录校验db DB() user db.query( SELECT * FROM users WHERE username%s AND password%s, (admin, 123456) ) if user: print(登录成功:, user) else: print(用户名或密码错误) db.close()✅参数化 SQL防止 SQL 注入五、Flask MySQL 登录接口实战重点 ✅1️⃣ 安装 Flaskpip install flask2️⃣ Flask 项目结构flask_mysql/ ├── app.py ├── db.py └── requirements.txt3️⃣ db.py数据库封装import pymysql def get_db(): return pymysql.connect( hostlocalhost, port3306, userroot, password你的密码, databaseflask_demo, charsetutf8mb4, cursorclasspymysql.cursors.DictCursor )4️⃣ app.py登录接口from flask import Flask, request, jsonify from db import get_db import pymysql app Flask(__name__) app.route(/login, methods[POST]) def login(): data request.get_json() username data.get(username) password data.get(password) db get_db() try: with db.cursor() as cursor: sql SELECT id, username FROM users WHERE username%s AND password%s cursor.execute(sql, (username, password)) user cursor.fetchone() finally: db.close() if user: return jsonify({ code: 0, msg: 登录成功, data: user }) else: return jsonify({ code: 1, msg: 用户名或密码错误 }) if __name__ __main__: app.run(debugTrue)5️⃣ 测试接口Postman / curlcurl -X POST http://127.0.0.1:5000/login \ -H Content-Type: application/json \ -d {username:admin,password:123456}六、SQLAlchemy ORM进阶 · 强烈推荐企业项目几乎都用 ORM安装pip install flask_sqlalchemy pymysql示例ORM 版from flask import Flask from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SQLALCHEMY_DATABASE_URI] ( mysqlpymysql://root:密码localhost/flask_demo?charsetutf8mb4 ) app.config[SQLALCHEMY_TRACK_MODIFICATIONS] False db SQLAlchemy(app) class User(db.Model): __tablename__ users id db.Column(db.Integer, primary_keyTrue) username db.Column(db.String(50), uniqueTrue) password db.Column(db.String(100)) with app.app_context(): db.create_all()✅不用写 SQL✅更安全✅更易维护七、常见错误 解决方案错误原因解决Access denied密码 / 用户错检查 MySQL 用户Cant connectMySQL 没启动net start mysqlcharset 报错未指定 utf8mb4加 charsetSQL 注入字符串拼 SQL用%s占位符中文乱码编码不一致utf8mb4八、安全与规范非常重要✅密码永远不要明文import hashlib hashlib.sha256(password.encode()).hexdigest()✅生产环境配置不写死app.config.from_envvar(APP_CONFIG)✅连接用完必须关闭✅SQL 必须参数化九、学习路线建议阶段内容第 1 步PyMySQL 原生 SQL第 2 步Flask MySQL API第 3 步SQLAlchemy ORM第 4 步事务 / 连接池第 5 步Redis MySQL十、一句话总结 ✅Python 连 MySQL 装驱动 → 建连接 → 写 SQL → 关连接​真实项目 Flask SQLAlchemy 参数化 SQL