記錄一下SQLAlchemy 2.0連接mysql數據庫的方法及簡單使用

環境及依賴

Python:3.8
mysql:8.3
Flask:3.0.3
SQLAlchemy:2.0.37
PyMySQL:1.1.1

使用步驟

1、創建引擎，鏈接到mysql

engine = create_engine('mysql+pymysql://{username}:{password}@{ip}:3306/{database_name}', echo=True)

以上的變量替換成自己的即可，比如你需要連接本地的test_db數據庫，用戶名為root，密碼為123456，那么寫法就是：

engine = create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/test_db', echo=True)

echo參數的作用是開啟sql輸出，以便調試

2、打開數據庫鏈接會話

session = sessionmaker(engine)

3、保證線程安全

db_session = scoped_session(session)

db_session對象在保證線程安全的同時也封裝了常用的數據庫操作，方便使用

4、獲取基類

Base = declarative_base()

5、創建數據庫映射對象類

假設當前有一個數據庫表名為user，表結構如下：

class User(Base):__table__ = Table('user', Base.metadata, =engine)

autoload_with開啟表結構的自動反射加載

測試

至此準備工作已經全完成，下面使用flask框架寫一個簡單的登錄接口來測試

import jsonfrom flask import Flask, request
from sqlalchemy import create_engine, Table
from sqlalchemy.orm import sessionmaker, scoped_session, declarative_baseapp = Flask(__name__)# 創建一個引擎，鏈接到mysql
engine = create_engine('mysql+pymysql://root:123456@127.0.0.1:3306/test_db', echo=True)
# 打開數據庫鏈接會話
session = sessionmaker(engine)
# 保證線程安全
db_session = scoped_session(session)
# 獲取基類
Base = declarative_base()class User(Base):__table__ = Table('user', Base.metadata, autoload_with=engine)@app.route('/', methods=['POST'])
def login():# 取到json請求數據request_data = json.loads(request.data)username = request_data['username']password = request_data['password']# 根據username、password查詢數據庫中是否有匹配對象result = db_session.query(User).filter_by(username=username, password=password).first()if result is None:return "登錄失敗"return "登錄成功"if __name__ == '__main__':app.run()

運行當前項目

 * Serving Flask app 'main'* Debug mode: off
WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead.* Running on http://127.0.0.1:5000

向接口發送請求

成功取到數據，登錄成功：

2025-01-31 23:03:09,852 INFO sqlalchemy.engine.Engine BEGIN (implicit)
2025-01-31 23:03:09,855 INFO sqlalchemy.engine.Engine SELECT user.user_id AS user_user_id, user.username AS user_username, user.password AS user_password, user.nickname AS user_nickname, user.picture AS user_picture, user.create_time AS user_create_time, user.update_time AS user_update_time, user.job AS user_job 
FROM user 
WHERE user.username = %(username_1)s AND user.password = %(password_1)s LIMIT %(param_1)s
2025-01-31 23:03:09,855 INFO sqlalchemy.engine.Engine [generated in 0.00022s] {'username_1': '黎震南', 'password_1': '62927', 'param_1': 1}
127.0.0.1 - - [31/Jan/2025 23:03:09] "POST / HTTP/1.1" 200 -