投稿
  1. 首页
  2. python

基于flask框架和分层分模块的思想，集成sqlalchemy进行数据库 *** 作

code • 2022-5-6 • python • 阅读 48

基于flask框架和分层分模块的思想，集成sqlalchemy进行数据库 *** 作,第1张

文章目录
  • flask集成sqlalchemy
    • 1.快速入门
      • 1.1.数据库准备
      • 1.2.安装对应相应的包
      • 1.3.入门代码app.py
      • 1.4.运行app.py
    • 2.抛出问题
      • 2.1.如何统一对ORM返回的数据做处理？
      • 2.2.如何对代码进行分模块处理？
      • 2.3.在蓝图中如何集成sqlalchemy
        • 2.3.1.config.py
        • 2.3.2.database.py
        • 2.3.3.app.py
        • 2.3.5.user蓝图
          • 2.3.5.1.controller.py
          • 2.3.5.1.model.py
          • 2.3.5.1.service.py

之前的文章中，我们学习了 Python发布微服务到注册中心Nacos
在这篇文章中，我们将集成一个ORM数据库框架，同时会引入分层分模块的思想。即在蓝图模块中集成sqlalchemy。重要的是思想，而不是具体API的用法， 本文重心分层分模块的思想，具体API使用，这里不做复述。最终效果，如下所示：

首先我们先跑个简单的例子。

flask集成sqlalchemy 1.快速入门 1.1.数据库准备
  1. 准备好数据库，并且数据库有数据，例如
1.2.安装对应相应的包 
pip install flask
pip install pymysql
pip install flask-sqlalchemy
1.3.入门代码app.py 
from flask.app import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://数据库用户名:数据库密码@数据库地址:3306/数据库名?charset=utf8'

db = SQLAlchemy(app)

@app.route("/user/info/all",methods=['get'])
def all_user():
    users = SysUser.query.all()
    data = []
    for user in users:
        temp = {'id':user.id,'user_name':user.user_name,'real_name':user.real_name,'phone':user.phone}
        data.append(temp)
    result = {"code":"200","msg":" *** 作成功","data":data}
    return result
#这里的SysUser替换成你数据库具体的表，属性字段也改成你数据库对应的字段
class SysUser(db.Model):
    __tablename__ = 'sys_user'
    id = db.Column(db.String, primary_key=True)
    user_name = db.Column(db.String())
    real_name = db.Column()
    phone = db.Column()
    org = db.Column()
    mail = db.Column()
    
if __name__ == "__main__":
    app.run(port=8080,debug=True)
1.4.运行app.py

启动项目后，接着我们通过postman访问该接口，或者浏览器访问


至此快速入门完毕，当然我看了，很多培训机构的老师也是这样入门的，接着就开始各种API的学习了，包括网上的很多人也是这样思路，研究了两天吧（本人学习python转过来也就3天，这里我就研究了2天），我就发现都是按照这样的套路写的。
听过很多人说，python就是一个脚本语言，不适合那种专业的程序员用的语言，适合那种不是程序员，但是需要写程序来实现一些功能，比如做算法的，做人工智能的，做自动化测试的。当我接触了周围使用python的人以后，我更加赞成了这个观点。因为我发现不仅是他们，同样的培训机构的老师在讲课的时候也是这样。

2.抛出问题

注：本文只从思想上去解决一些问题，至于那种API使用，我就不啰嗦了。

2.1.如何统一对ORM返回的数据做处理？

观察如下这快代码，我希望返回Json数据给调用我的人，但是ORM返回的是对象或者数组（数组中的元素数对象），但是我们返回给接口调用者，需要的是json。问题来了，我返回一个两字段，你让我写代码也就算了，要是我接口返回十几个二十个字段，甚至上百个字段，你也要让我，挨个手写？显然不可能，那样工作量，太大了，而且代码太难维护了。

@app.route("/user/info/all",methods=['get'])
def all_user():
    users = SysUser.query.all()
    data = []
    for user in users:
        temp = {'id':user.id,'user_name':user.user_name,'real_name':user.real_name,'phone':user.phone}
        data.append(temp)
    result = {"code":"200","msg":" *** 作成功","data":data}
    return result

Spring体系中，各种MVC框架都给我们准备好各种解析器，因此，我们无论返回的是何种数据类型，都可以给我们格式化成JSON，但是python我并没有找到，我找了很多资料也没有找到，返回的数据都是如下这样，自己构建一个字典，然后返回字典,或者数组。

而在java中，我们都是定义如下的类，然后无论是什么数据，都可以放进去，并且最终解析器给我们解析成Json

尝试了python中，很多json解析，发现都不好用，因此我做了如下一个简单工具类copy_utils和R

class copy_utils:
    
    @staticmethod
    def obj_to_dic(obj):
        '''
        将传入的data对象转成字典
        '''
        result = {}
        for temp in obj.__dict__:
            if temp.startswith('_') or temp == 'metadata':
                continue
            result[temp] = getattr(obj, temp)
        return result
    
    @staticmethod
    def obj_to_list(list_obj):
        '''
        将传入的data对象转成List,list中的元素是字典
        '''
        result = []
        for obj in list_obj:
            result.append(copy_utils.obj_to_dic(obj))
        return result
        
class R(object):
    
    @staticmethod
    def ok(data):
        result = {"code":"200","msg":" *** 作成功","data":data}
        return jsonify(result)
    
    @staticmethod
    def erro(code = 500,msg = "系统异常"):
        result = {"code":code,"msg":msg}
        return jsonify(result)

2.2.如何对代码进行分模块处理？

这块其实问题还不大，因为很多培训机构还是讲了蓝图。
这里我就大概提一下。我们项目往往不是一个人开发，而且项目肯定不止一个模块，好的状态是，不同的人负责不同的模块，各个模块在不同的项目（包）下。蓝图的作用就是让我们的Flask项目更加模块化，结构更加清晰,为了更好的管理项目 让项目达到分层解耦 而产生的。可以将相同模块的视图函数放在同一个蓝图下，同一个文件中，方便管理。

2.3.在蓝图中如何集成sqlalchemy

但是…我特码的惊呆，我看了很多人写的博客，或者培训机构的视屏，因为很多人压根没有把sqlalchemy和蓝图结合起来一起用。我整个学习Python用了3天，这个问题我就占据了1天多。
于是能看到，所有和数据库表对应的python对象，全特码的在这个app.py文件中，所有查询数据库的 *** 作，全特码的也是在这个app.py文件。我算是明白了，为啥我们项目以及我见过的很多写python的，能把python项目的入口文件（本文中app.py）写几千行，甚至上万行。路由在入口文件，业务处理，也在入口文件，甚至数据库增删查该都在入口文件，简直服了。
当我开始创建不同的包，存放不同业务的代码。

controller主要是用来存放对外暴露的接口
service主要是用来存放处理业务逻辑的
model主要是用来存放数据库对象的
但是，我发现，当我在service中，把db引入过去的时候，反正我编译的时候没报错，运行的时候就报错了

2.3.1.config.py

因此对app开始进行一些更改，将数据库配置相关的信息，写入到config.py

#数据库连接配置， 参数说明 方言+数据库驱动：//用户名：密码@数据库地址:/3306/数据库
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://root:<密码>@<地址>:3306/<数据库名>?charset=utf8'
#关闭数据库修改跟踪 *** 作[提高性能]，可以设置为True，这样可以跟踪 *** 作：
SQLALCHEMY_TRACK_MODIFICATIONS = False
#开启SQL语句打印
SQLALCHEMY_ECHO = True
#开启数据库的自动提交功能[一般不使用]
SQLALCHEMY_COMMIT_ON_TEARDOWN = False
#连接池大小，默认为5
SQLALCHEMY_POOL_SIZE = 5
#连接池超时时间，默认为10
SQLALCHEMY_POOL_TIMEOUT = 10
#多少秒后自动回收连接。这对 MySQL 是必要的， 它默认移除闲置多于 8 小时的连接。注意如果 使用了 MySQL ， Flask-SQLALchemy 自动设定 这个值为 2 小时。
#SQLALCHEMY_POOL_RECYCLE =
2.3.2.database.py 
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
2.3.3.app.py 
from flask.app import Flask
from com.hutao import config
from com.hutao.database import db
from com.hutao.flaskdemo.org.controller import org
from com.hutao.flaskdemo.user.controller import user

app = Flask(__name__)

app.register_blueprint(user, url_prefix='/user')
app.register_blueprint(org, url_prefix='/org')
#从config读取
app.config.from_object(config)
#数据库初始化
db.init_app(app)

if __name__ == "__main__":
    app.run(port=8080,debug=True)
2.3.5.user蓝图 2.3.5.1.controller.py 
from flask import Blueprint

from com.hutao.flaskdemo.base.R import R

org = Blueprint("org", __name__)

@org.route("/info/all",methods=['get'])
def info_all():
    return R.ok(data=None)
2.3.5.1.model.py 
from com.hutao.database import db
class SysUser(db.Model):
    __tablename__ = 'sys_user'
    id = db.Column(db.String, primary_key=True)
    user_name = db.Column(db.String())
    real_name = db.Column()
    phone = db.Column()
    org = db.Column()
    mail = db.Column()
    create_time = db.Column()
2.3.5.1.service.py 
from com.hutao.flaskdemo.base.utils import copy_utils
from com.hutao.flaskdemo.user.model import SysUser
class user_service(object):
    
    @staticmethod
    def query_userinfo():
        #查询所有
        users = SysUser.query.all()
        return copy_utils.obj_to_list(users)

完整代码资源（0积分下载）：https://download.csdn.net/download/m0_37892044/85237529

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/langs/800189.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
code code 管理员组
0 0
上一篇 2022-05-06
下一篇 2022-05-06

发表评论

登录后才能评论

评论列表（0条）

保存