Flask框架（四）通过数据模型名称访问模型

本文介绍Flask中，通过数据模型名称访问数据模型，下面详细介绍：

1、定义数据模型基类

定义数据模型基类BaseModel，数据模型继承这个基类

from flask_sqlalchemy import SQLAlchemy import logging import json # 客户数据库连接，用作数据模型 # 此处db是默认的数据库引擎连接SQLALCHEMY_DATABASE_URI，指向数据库根目录，运行sql语句时要指定具体数据库 db = SQLAlchemy() class BaseModel(object): def update(self): # 更新记录 try: db.session.commit() # 事务提交 except Exception as e: db.session.rollback() # 事务回滚 logging.error('[BaseModel->update]Failed to run update, {0}'.format(e)) return False return True def add_update(self): # 增加一条记录 try: db.session.add(self) # 添加数据对象 db.session.commit() # 事务提交 except Exception as e: db.session.rollback() # 事务回滚 logging.error('[BaseModel->add_update]Failed to run add_update, {0}'.format(e)) return False return True def add_all_update(self, datas: list): # 增加多条记录 try: db.session.add_all(datas) # 添加数据对象 db.session.commit() # 事务提交 except Exception as e: db.session.rollback() # 事务回滚 logging.error('[BaseModel->add_all_update]Failed to run add_all_update, {0}'.format(e)) return False return True def delete(self): # 删除一条记录 try: db.session.delete(self) # 删除数据对象 db.session.commit() # 事务提交 except Exception as e: db.session.rollback() # 事务回滚 logging.error('[BaseModel->delete]Failed to run delete, {0}'.format(e)) return False return True def to_json(self): # SQLAlchemy转json对象 fields = {} for field in [x for x in self.__dict__ if not x.startswith('_') and x != 'metadata']: data = self.__getattribute__(field) try: # 非json识别的数据类型在这里转换，这里以DataTime类型转换举例 if isinstance(data, datetime): data = data.strftime('%Y-%m-%d %H:%M:%S') json.dumps(data) # this will fail on non-encodable values, like other classes fields[field] = data except TypeError: fields[field] = None logging.error(traceback.format_exc()) return fields # 执行自定义sql语句查询，失败最多重试3次 def mysql_execute(sql: str): results = [] session = db.session() n = 0 while (not results and n < 3): n += 1 try: resultproxy = session.execute(sql) if resultproxy.keys(): results = resultproxy.fetchall() except Exception as e: session.rollback() # 事务回滚 logging.error('[mysql_execute]Failed to run mysql_execute, {0}'.format(e)) session.close() return results

创建数据库函数

def create_db(dbname: str): if not mysql_execute("show databases like '{0}';".format(dbname)): sql = 'CREATE DATABASE IF NOT EXISTS {0} default charset utf8 COLLATE utf8_general_ci;'.format(dbname) mysql_execute(sql)

生成数据表函数

def create_db_client(dbname: str): try: create_db(dbname) # 创建数据库 db.create_all() # 创建客户数据表 except Exception as e: logging.error('[create_db_client]Failed to run create_db_client, {0}'.format(e))

2、定义数据模型

这里以定义用户信息数据模型举例，代码如下：

from FlaskDemo.controls.Base import BaseModel, db import logging from FlaskDemo.status_code import RET, ret_map from flask import jsonify from werkzeug.security import generate_password_hash from FlaskDemo.controls.AccountPost import AccountPost from FlaskDemo.controls.AccountDpart import AccountDpart import json import re from sqlalchemy import TIMESTAMP, func # 客户用户信息数据表模型 class AccountUsers(BaseModel, db.Model): __bind_key__ = 'client' # 数据库引擎，这就表示数据库名称，在config.py中定义 __tablename__ = 'account_users' # 账户信息->用户表 Id = db.Column(db.INTEGER, primary_key=True) Name = db.Column(db.String(50), unique=True) # 用户名称 Pwd = db.Column(db.String(300)) # 用户密码 Nick = db.Column(db.String(150)) # 用户姓名 PostId = db.Column(db.INTEGER, nullable=False) # 职位ID Mobile = db.Column(db.String(150)) # 用户手机号码 Email = db.Column(db.String(150)) # 用户邮箱地址 RoleIds = db.Column(db.String(255)) # 用户角色ID，json数组 GroupIds = db.Column(db.String(255)) # 用户组ID，json数组 Authview = db.Column(db.Text, default='{}') # 用户查看权限 Authoper = db.Column(db.Text, default='{}') # 用户操作权限 Valid = db.Column(db.INTEGER, default=1) # 账号是否有效：0注销，1生效 RegTime = db.Column(TIMESTAMP, server_default=func.current_timestamp(), index=True) # 检验限制创建时间

3、获取所有数据模型信息

获取所有数据模型字典，key为模型名称，value为模型

def get_type_query(): typedict = {} for clazz in db.Model._decl_class_registry.values(): try: typedict[clazz.__name__] = clazz except Exception as e: logging.error(e) return typedict

4、访问例子

def handle_requests(qrytype: str, qryfunc: str, argc, Id: int = None): dbClass = get_type_query().get(qrytype) if not dbClass: return jsonify(code=RET.PARAMERR, msg=u'查询类型错误') dbcls = dbClass() if isinstance(argc, str): try: argcjson = json.loads(argc) except Exception as e: logging.error('[handle_requests]Failed to json.load, {0}'.format(e)) logging.error(traceback.format_exc()) return jsonify(code=RET.PARAMERR, msg='Failed to json.load, {0}'.format(e)) elif isinstance(argc, dict): argcjson = argc else: return jsonify(code=RET.PARAMERR, msg=u'argc查询类型错误') # print(dir(dbcls)) if qryfunc in dir(dbcls): if Id: value = dbcls.__getattribute__(qryfunc)(argcjson, Id) else: value = dbcls.__getattribute__(qryfunc)(argcjson) return value else: return jsonify(code=RET.PARAMERR, msg=qryfunc + '方法不存在，请联系开发人员')

说明：

handle_requests函数功能：通过传入qrytype（模型名称）、qryfunc（模型里面定义的方法），argc（模型方法传入的参数）参数，实现调用模型里面的方法get_type_query().get(qrytype)：获取模型dbcls.getattribute(qryfunc)(argcjson)：调用模型里面方法

5、完整教程地址

Flask搭建项目完整教程：Flask+Vue搭建系统