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) -
