Flask Web开发入门（七）之SQLAlchemy

缺乏、安全感 2022-06-02 09:16 246阅读 0赞

[上一章][Link 1]中，我们简单介绍了使用 **SQLAlchemy** **ORM**框架来访问MySQL数据库，本章我们将继续就此话题进行展开。

> SQLAlchemy is the Python SQL toolkit and Object Relational Mapper that gives application developers the full power and flexibility of SQL.

SQLAlchemy包含了两个模块：**[SQLAlchemy ORM][]** 和**[SQLAlchemy Core][SQLAlchemy ORM]**，简单来讲：

*  **Core**是一种构建在表达式语言之上的一种API，一种SQL抽象工具包，允许用户以Python的语法格式构建SQL表达式，
 *  而**ORM** 基于**Core**之上，一种更高级的、更加抽象的一种使用模式，其提供了现实业务对象与关系型数据库中表的一种关联关系，并实现两者之间的轻松转换。

**Essential SQLAlchemy (Jason Myers, O’Reilly)**描述如下：

*  If you have data for which business objects are not needed, use Core.
 *  If you view your data as business objects, use ORM.

另外，既然我们使用Flask作为Web开发框架，同时使用**SQLAlchemy**作为后台数据库连接中间件，那么我们不得不提**[Flask-SQLAlchemy][]**。

> Flask-SQLAlchemy is an extension for Flask that adds support for SQLAlchemy to your application. It requires SQLAlchemy 0.8 or higher. It aims to simplify using SQLAlchemy with Flask by providing useful defaults and extra helpers that make it easier to accomplish common tasks.

**[Flask-SQLAlchemy][]**是一个Flask插件，支持**SQLAlchemy**（要求**SQLAlchemy**版本0.8以上），**[Flask-SQLAlchemy][]**一个重要的功能实现是提供了一个完整的分页封装实现。

下面，我们示例说明各自用法，首先定义一个公共方法，用以获取数据库连接：

>  get\_connection\_with\_url

# get connection using url
    def get_connection_with_url(url):
        try:
            engine = create_engine(url, echo=True)
            conn = engine.connect()
            return conn
        except Exception as e:
            logger.debug("Exception is %s" % e)
            return None

#### **SQLAlchemy ORM** ####

*  定义类**Monitor**

Base = declarative_base()
    # http://flask-sqlalchemy.pocoo.org/2.3/
    # http://docs.sqlalchemy.org/en/latest/
    # SQLAlchemy orm
    class Monitor(Base):
        __tablename__ = 't_monitor'
    
        id = Column('id', Integer, primary_key=True)
        credit_type = Column('credit_type', String(128))
        query_type = Column('query_type', String(128))
        credit_status = Column('credit_status', String(128))
        monitor_time = Column('monitor_time', String(128))
        elapsed_time = Column('elapsed_time', String(128))
        create_time = Column('create_time', String(128))
    
        def __init__(self, id, credit_type, query_type, credit_status, monitor_time, elapsed_time, create_time):
            self.id = id
            self.credit_type = credit_type
            self.query_type = query_type
            self.credit_status = credit_status
            self.monitor_time = monitor_time
            self.elapsed_time = elapsed_time
            self.create_time = create_time
    
        def __repr__(self):
            return '<id is %s, creditType is %s, queryType is %s, creditStatus is %s, monitorTime is %s, elapsedTime is %s>' % (
                self.id, self.credit_type, self.query_type, self.credit_status, self.monitor_time, self.elapsed_time)

*  获取表记录信息

# http://docs.sqlalchemy.org/en/latest/
    # SQLAlchemy orm
    def get_monitor_with_orm():
        s = flask_db.get_connection_session(url)
        print(s.query(Monitor).limit(2).all())
        print(s.query(Monitor).first())
        print(type(s.query(Monitor)))
        print(s.query(Monitor).count())

#### **SQLAlchemy Core** ####

*  定义表对象

# SQLAlchemy core
    metadata = MetaData()
    monitor = Table('t_monitor', metadata, Column('id', Integer, primary_key=True)
                    , Column('credit_type', String(128))
                    , Column('query_type', String(128))
                    , Column('credit_status', String(128))
                    , Column('monitor_time', String(128))
                    , Column('elapsed_time', String(128))
                    , Column('create_time', String(128)))

*  获取表记录信息

# SQLAlchemy core
    def get_monitor_with_core():
        conn = flask_db.get_connection_with_url(url)
        sql = select([monitor])
        result = conn.execute(sql)
        print(result.rowcount)
        print(type(result.fetchall()))

#### **Flask-SQLAlchemy** ####

*  定义FlaskMonitor类

# Flask-SQLAlchemy
    class FlaskMonitor(db.Model):
        __tablename__ = 't_monitor'
    
        id = Column('id', Integer, primary_key=True)
        credit_type = Column('credit_type', String(128))
        query_type = Column('query_type', String(128))
        credit_status = Column('credit_status', String(128))
        monitor_time = Column('monitor_time', String(128))
        elapsed_time = Column('elapsed_time', String(128))
        create_time = Column('create_time', String(128))
    
        def __init__(self, id, credit_type, query_type, credit_status, monitor_time, elapsed_time, create_time):
            self.id = id
            self.credit_type = credit_type
            self.query_type = query_type
            self.credit_status = credit_status
            self.monitor_time = monitor_time
            self.elapsed_time = elapsed_time
            self.create_time = create_time
    
        def __repr__(self):
            return '<id is %s, creditType is %s, queryType is %s, creditStatus is %s, monitorTime is %s, elapsedTime is %s>' % (
                self.id, self.credit_type, self.query_type, self.credit_status, self.monitor_time, self.elapsed_time)

*  获取表记录信息，注意我们这里调用了**paginate**方法实现了分页功能

# using flask_sqlalchemy
    def get_monitor_flask_sqlalchemy(page=1, limit=10):
        try:
            logger.debug('get_monitor_flask_sqlalchemy: page is %s, limit is %s' % (page, limit))
            return FlaskMonitor.query.paginate(page, limit)
        except Exception as e:
            logger.debug("Exception in get_monitor_flask_sqlalchemy %s" % e)
            return None

以上，简单示例说明了**SQLAlchemy**相关模块的用法，下面，我们使用**[Flask-SQLAlchemy][]**示例演示前台列表分页展示功能，前台列表我们没有使用Jinjia模块，而使用了Layui，页面代码如下：

<!DOCTYPE html> <html lang="zh"> <head> <meta charset="UTF-8"> <meta http-equiv="X-UA-Compatible" content="IE=edge,chrome=1"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>首页</title> <link rel="stylesheet" type="text/css" href="{ { url_for('static', filename='js/layui/css/layui.css') }}"/> <script src="{ { url_for('static', filename='js/jquery-2.1.1.min.js') }}" type="text/javascript"></script> <script src="{ { url_for('static', filename='js/layui/layui.js') }}" type="text/javascript"></script> <link rel="shortcut icon" href="{ { url_for('static', filename='images/wd_favicon.ico')}}"/> </head> <body> <div style="width: 80%; float:left; margin: 10px 5px;font-family: Consolas;font-size: 20px;"> {% if name %} Hello { { name }}! {% else %} Hello World! {% endif %} </div> <div style="float:right; width: 10%;margin: 10px 5px;font-family: Consolas;font-size: 20px;"> <a href="#" onclick="javascript:logout();">logout</a> </div> <table class="layui-table" id="list"></table> <script type="text/javascript"> var logout = function () { window.location.href = '/logout'; } </script> <script> layui.use('table', function () { var table = layui.table; table.render({ elem: '#list' , url: '/api/list' , cellMinWidth: 80 , limit: 5 , page: true , cols: [[ {field: 'id', title: 'ID', sort: true, align: 'center'} , {field: 'credit_type', title: '认证类型', align: 'center'} , {field: 'query_type', title: '请求类型', sort: true, align: 'center'} , {field: 'credit_status', title: '请求状态', align: 'center'} , {field: 'monitor_time', title: '监控时间', align: 'center'} , {field: 'elapsed_time', title: '监控耗时', align: 'center'} , {field: 'create_time', title: '创建时间', sort: true, align: 'center'} ]] }); }); </script> </body> </html>

后台JSON数据接口实现如下，代码涉及了Python 类的序列化，[后续章节][Link 2]我们展开：

@app.route('/api/list', methods=['GET'])
    def get_list():
        page = request.args.get('page')
        limit = request.args.get('limit')
        logger.debug("get_list: page = %s, limit = %s" % (page, limit))
        pages = flask_monitor.get_monitor_flask_sqlalchemy(int(page), int(limit))
        if pages is None:
            return jsonify({
       "code": 0, "msg": "", "count": 0, "data": {}})
        else:
            data = []
            for item in pages.items:
                item.__dict__['_sa_instance_state'] = ''
                item.__dict__['create_time'] = "%s" % item.__dict__['create_time']
                item.__dict__['monitor_time'] = "%s" % item.__dict__['monitor_time']
                data.append(item.__dict__)
            result = json.dumps({
       "code": 0, "msg": "", "count": pages.total, "data": data})
            return result

实现效果如下：  
![这里写图片描述][SouthEast]

**源码参考：**[https://github.com/ypmc/flask-sqlalchemy-web][https_github.com_ypmc_flask-sqlalchemy-web]

[Link 1]: http://blog.csdn.net/kangkanglou/article/details/79007358
[SQLAlchemy ORM]: http://docs.sqlalchemy.org/en/latest/
[Flask-SQLAlchemy]: http://flask-sqlalchemy.pocoo.org/2.3/
[Link 2]: http://blog.csdn.net/kangkanglou/article/details/79038223
[SouthEast]: /images/20220602/417c389b3d0145b4ad739e8c550d81bb.png
[https_github.com_ypmc_flask-sqlalchemy-web]: https://github.com/ypmc/flask-sqlalchemy-web