Python

• 说明

• python作为一种解释性语言，有着其独特的优势。有时候我们会用python写一些简单的http接口或者脚本来提供一些服务，以下介绍下 python操作mysql的一款ORM框架SQLAlchemy 的使用。

• 首先 我们先建立个配置类，用来做一些配置性的东西，类似于java的properties文件，只不过这里把他作为一个类来管理起来

• 代码样例

  • mysql的配置类

  class BaseConfig():
    #以下是我个人常用的配置类，仅做参考
    #Mysql 配置
    SQLALCHEMY_DATABASE_URI="mysql+pymysql://user:passwd@**.**.***.**:3306/dataBaseName"
    SQLALCHEMY_TRACK_MODIFICATIONS = False #如果设置成 True (默认情况)，Flask-SQLAlchemy 将会追踪对象的修改并且发送信号。这需要额外的内存， 如果不必要的可以禁用它
    #以下配置不常用
    SQLALCHEMY_POOL_SIZE = 5 #数据库连接池的大小。默认是数据库引擎的默认值 （通常是 5）。
    SQLALCHEMY_POOL_TIMEOUT = 10 #指定数据库连接池的超时时间。默认是 10
    SQLALCHEMY_MAX_OVERFLOW = 2 #控制在连接池达到最大值后可以创建的连接数。当这些额外的 连接回收到连接池后将会被断开和抛弃
    SQLALCHEMY_POOL_RECYCLE = 2 #动回收连接的秒数。这对 MySQL 是必须的，默认 情况下 MySQL 会自动移除闲置 8 小时或者以上的连接。 需要注意地是如果使用 MySQL 的话，Flask-SQLAlchemy 会自动地设置这个值为2小时
    #Mongo配置
    MONGODB_PORT = 27017
    MONGODB_HOST = "**.**.**"
    MONGODB_DB = "dbName"
    MONGODB_USERNAME = "dbuser"
    MONGODB_PASSWORD = "dbpasswd"
    #MQ配置 rabbitmq
    MQ_USER_NAME = 'name'
    MQ_USER_PAWD = 'pwd'
    MQ_URL = 'host'
    MQ_HOST = 5672
    #初始化app
    app = Flask('dbclient') 因为我们做的是web服务所以这里用到了python的web框架         FLask 有兴趣的可以了解下
    #初始化mysqldb
    app.config.from_object(BaseConfig)  这里用我们的配置类去初始化mysql连接
    db = SQLAlchemy(app)
  

• 建立两个数据库映射的实体类

  • 用户表

  from config.Config import db  #这里我们引用配置类中初始化好的db
     
    #建立数据库和实体类的映射
    class User(db.Model):
        #表名
        __tablename__ = 'user_info' 
        #字段属性
        id = db.Column(db.Integer, primary_key=True)
        user_name = db.Column(db.String(100))
        user_pswd = db.Column(db.String(100))
        user_token = db.Column(db.String(100))
        usercname = db.Column(db.String(100))
     
        #__init__方法为python类的构造方法
        def __init__(self,user_name, user_pswd, user_token,usercname):  # 这里实现以下构造，将其作为插入新值的方法
            print("User 构成函数被调用")
            # self.id = id
            self.user_name = user_name
            self.user_pswd = user_pswd
            self.user_token = user_token
            self.usercname = usercname
     
        #这个方法作为对象的打印方法，类似于java类中 自己重写的 toString()方法，实现这个方法后，print(类) 将会执行这个方法进行打印
        def __repr__(self):
            # 输出对象
            return '<User %i %r %r %r %r>' % (self.id,self.user_name,self.user_token,self.user_pswd,self.usercname)
  

  • token表

      from config.Config import db
    class Token(db.Model):
        __tablename__ = 'token_info'
        id = db.Column(db.Integer, primary_key=True)
        token_id = db.Column(db.String(100))
        user_cname = db.Column(db.String(100))
        user_id = db.Column(db.String(100))
     
        def __init__(self, token_id, user_cname,user_id):  # 插入新值的方法
            print("Token构成函数被调用")
            # self.id = id
            self.token_id = token_id
            self.user_cname = user_cname
            self.user_id = user_id
     
        def __repr__(self):
            # 输出对象
            return '<Token %i %r %r %r>' % (self.id,self.token_id,self.user_cname,self.user_id)
  

• 建立数据库操作类，主要用于封装数据库操作的方法。

  • 代码如下：

      class UserInfo():
        # 全表查询
        def findAll(self):
            return User.query.all()
        #根据ID主键查询
        def getUserInfoById(self,id):
            return User.query.get(id)
        #自定义查询条件
        def getUserInfoByUserName(self,userName):
            return  User.query.filter_by(usercname = userName).all()
     
    class TokenInfo():
        def findAll(self):
            return Token.query.all()
        def getTokenInfoById(self,id):
            return Token.query.get(id)
     
    #复杂查询以及多表关联查询DEMO展示
    class UtilsQuery():
        @timelog    #这里的timelog是我自己的写的装饰器，目前实现的功能类似于java的aop切面，对方法进行环绕增强，在方法执行前后计算方法的执行时间用
        def find(self):
            #这里通过关联User和Token两张表进行关联查询
            return User.query.filter(User.id == Token.user_id).filter(Token.user_cname == 'XXXX').all()
            # return db.session.query(User).filter(User.id == Token.user_id).filter(Token.user_cname=='XXXX').all()
        @timelog
        def findByPage(self,pageNum,size):
            #这里展示分页查询
            if pageNum < 1:
                pageNum = 1
            if size < 1:
                size = 2
            start = (pageNum - 1)*size
            end = pageNum*size
            return User.query.slice(start,end).all()
  

• 建立controller层，提供web调用

  • DEMO代码如下：

    @app.route('/controllertest',methods=['GET','POST'])
    def hello_world():
        print('接受到请求')
        returnData = {}
        returnData['data'] = "success"
     
        userInfoDto = UserInfo()
        print('\nid查询')
        print(userInfoDto.getUserInfoById(1))
        print('\n条件查询')
        print(userInfoDto.getUserInfoByUserName("XXXX"))
        print('\n全表查询')
        print(TokenInfo().findAll())
        print('\n分页查询')
        print(UtilsQuery().findByPage(1,2))
        return json.dumps(returnData, ensure_ascii=False)