Flask笔记三之连接使用数据库

本文首发于公众号：Hunter后端

原文链接：Flask笔记三之连接使用数据库

这一节介绍 Flask 与数据库的连接，以及接口里查询数据的操作。

这里使用的是 SQLAlchemy + pymysql 实现与数据库的连接，SQLAlchemy 的详细介绍见之前的笔记有过一篇基础的介绍，这里只介绍与 Flask 的嵌入使用。

本篇笔记的代码都已经提交到 github 上，可使用下面的操作获取代码：

git clone https://github.com/x1204604036/flask_backend.git

以下是本篇笔记目录：

安装依赖
定义和连接数据库
接口使用数据库

0、安装依赖

安装相关依赖：

pip3 install sqlalchemy==2.0.19
pip3 install pymysql==1.1.0

1、定义和连接数据库

首先，在 app/config/development.py 里定义我们的数据库相关变量，内容如下：

# app/config/development.pyMYSQL_CONNECT_TYPE = "mysql+pymysql"
MYSQL_USERNAME = "root"
MYSQL_PASSWORD = "123456"
MYSQL_HOST = "192.168.1.3"
MYSQL_PORT = 3306
MYSQL_DB_NAME = "flask_db"

接下来定义 user 相关的 model，在 app/ 文件夹下创建一个 models/ 文件夹，其下创建一个 user.py 文件，内容如下：

# app/models/user.pyfrom sqlalchemy import Column, Integer, String
from app.database import Baseclass User(Base):__tablename__ = "user"id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String(50), unique=True)email = Column(String(120), default="", comment="邮箱")password = Column(String(120))

在 app/ 文件夹下创建一个文件 database.py，其内容如下：

# app/database.py from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from app.config import configengine_str = "%s://%s:%s@%s:%s/%s?charset=utf8" % (config.MYSQL_CONNECT_TYPE,config.MYSQL_USERNAME,config.MYSQL_PASSWORD,config.MYSQL_HOST,config.MYSQL_PORT,config.MYSQL_DB_NAME,
)engine = create_engine(engine_str)db_session = scoped_session(sessionmaker(autocommit=False,autoflush=False,bind=engine,)
)Base = declarative_base()
Base.query = db_session.query_property()def init_db(app):import app.models.userBase.metadata.create_all(bind=engine)

在这里，从 config 配置文件中获取 MySQL 的连接信息，然后创建 engine

最后定义一个初始化数据库的函数，用于创建所有定义了的数据库模型，在这里是引入了 User 这个 model

然后在 app/init.py 里引入数据库，文件内容如下：

# app/__init__.pyfrom flask import Flaskdef create_app():app = Flask(__name__)app.config.from_object("app.config.config")app.secret_key = b'f0e97d3012eed1c2939ac1a62ce1e8d455e86fa9da47b26c94a9af4119be29d6'from . import databasedatabase.init_db(app)from .user import bp_userapp.register_blueprint(bp_user.bp)return app

启动 Flask 服务，运行 flask run，就可以看到对应的数据库下多了我们创建的 user 表了。

接下来，我们改写上一篇笔记中的接口定义，从数据库中获取数据

2、接口使用数据库

以下是 app/user/bp_user.py 的内容：

from flask import Blueprint, request, session, jsonify
from werkzeug.security import generate_password_hash, check_password_hash
from app.models.user import User
from app.database import db_sessionbp = Blueprint("user", __name__, url_prefix="/user")@bp.route("/register", methods=("POST",))
def register():request_json = request.jsonusername = request_json.get("username")password = request_json.get("password")email = request_json.get("email")if not username or not password or not email:return {"code": -1, "msg": "请传递用户名、密码、邮件等必要信息"}user = User.query.filter(User.username == username, User.email == email).first()if user:return {"code": -1, "msg": "用户已存在"}new_user = User(username=username,email=email,password=generate_password_hash(password))db_session.add(new_user)db_session.commit()return {"code": 0, "msg": "注册成功"}@bp.route("/login", methods=("POST",))
def login():username = request.form.get("username")password = request.form.get("password")if not username or not password:return {"code": -1, "msg": "请上传用户名和密码"}user = User.query.filter(User.username == username).first()if not user:return {"code": -1, "msg": "用户不存在"}if not check_password_hash(user.password, password):return {"code": -1, "msg": "密码验证失败"}session.clear()session["user_id"] = user.idreturn {"code": 0, "msg": "success"}def login_required(func):def wrapped(*args, **kwargs):if session.get("user_id") is None:return jsonify({"code": -1, "msg": "请先登录系统"}), 401return func(*args, **kwargs)return wrapped@bp.route("/logout", methods=("POST",))
def logout():session.clear()return {"code": 0, "msg": "logout success"}@bp.route("/user_info", methods=("POST",))
@login_required
def get_user_info():user_id = session.get("user_id")user_info = {"user_id": user_id}return {"code": 0, "msg": "success", "user_info": user_info}