Sqlalchemy使用：连接Mysql/Pgsql，并用Orm和原生Sql增删改查

港控/mmm° 2023-10-04 17:44 23阅读 0赞

注意创建表的时候，Pgsql不要创建User表，因为User表是Pgsql的关键字，会出错。

#### Flask-SQLAlchemy ####

Flask-SQLAlchemy库让flask更方便的使用SQLALchemy,是一个强大的关系形数据库框架,既可以使用orm方式操作数据库,也可以使用原始的SQL命令.

#### 需要安装的包 ####

MySQL

SQLAlchemy

Flask-SQLAlchemy

mysqlclient

#### 配置Flask-SQLAlchemy ####

示例：

MySQL: mysql://username:password@hostname/database

PgSql：postgresql://用户名:密码@localhost:8891/pgdbname

![7be4d6358abe44ba96c2b1c6cc1f3cd3.png][]

数据库创建和配置：

![fa2ed6dc96a5410bb14acd3c48fb8a48.png][]

create table Users
    (
        id serial primary key,
        nick_name varchar(255)
    );

随便增加一条数据就可以了。

from sqlalchemy import Column, Integer, MetaData, VARCHAR
    from flask_sqlalchemy import SQLAlchemy
    from flask import Flask
    
    app = Flask(__name__)
    
    #配置数据库地址Mysql：
    app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://root:wjjf2008@127.0.0.1:3306/internet_plus?charset=utf8'
    
    #Pgsql配置： SQLALCHEMY_DATABASE_URI = "postgresql://postgres:postgres@localhost:8891/pgdbname"
    #该配置为True,则每次请求结束都会自动commit数据库的变动
    app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN'] = True
    
    db = SQLAlchemy(app)
    #也可以db = SQLAlchemy()        db.init_app(app)
    
    
    
    class User(db.Model):
        """用户"""
        __tablename__ = "User"
        id = db.Column(db.Integer, primary_key=True)  # 用户编号
        nick_name = db.Column(db.String(32), unique=True, nullable=False)  # 用户昵称
    
        def __repr__(self):
            """
            格式化输出
            """
            return '<User %r>' % self.nick_name
    
        def to_json(self):
            """
            为了转json提供的方法
            """
            dict = self.__dict__
            if "_sa_instance_state" in dict:
                del dict["_sa_instance_state"]
            return dict
    
    @app.route("/add_user")
    def add_user():
        # 添加测试数据库
        u = User()
        u.nick_name = "wang"
        # 将用户添加到数据库会话中
        db.session.add(u)
        # 将数据库会话中的变动提交到数据库中,如果不Commit,数据库中是没有改动的
        db.session.commit()
        return "添加成功"
    
    
    @app.route("/query_user")
    def query_user():
        # 返回所有用户保存到list中
        user_list = User.query.all()
        print(user_list)
        # 转成Json返回
        result = [u.to_json() for u in user_list]
        return jsonify(result), 200
    
    
    @app.route("/exec_sql")
    def exec_sql():
        # 执行原生SQL
        res = db.session.execute('select * from "User";').fetchall()
        emp_json_list = [dict(zip(item.keys(), item)) for item in res]
        print(emp_json_list)
        return jsonify(emp_json_list), 200
    
    
    if __name__ == '__main__':
        # db.create_all() #创建数据库
        app.run(host="0.0.0.0", port=5000)

直接访问：

![6ad602ab3568465fa483ee449bacf638.png][]

[7be4d6358abe44ba96c2b1c6cc1f3cd3.png]: https://img-blog.csdnimg.cn/7be4d6358abe44ba96c2b1c6cc1f3cd3.png
[fa2ed6dc96a5410bb14acd3c48fb8a48.png]: https://img-blog.csdnimg.cn/fa2ed6dc96a5410bb14acd3c48fb8a48.png
[6ad602ab3568465fa483ee449bacf638.png]: https://img-blog.csdnimg.cn/6ad602ab3568465fa483ee449bacf638.png