py_mybatis

• python的mybatis实现
• python的SqlTemplate实现

项目起源

特别鸣谢:mybatis-mapper2sql

借助这位大佬使我快速实现了py_mybatis,mapper xml解析主体代码都是 hhyo大佬的

本人是位javaer,刚刚开始学习python,当开始学习数据库相关api时,发现python中并未有很好的开发包,

重型orm 直接淘汰,不要问为什么,问就是我喜欢写sql,虽然sql写的不咋的

急需一种轻型的,开箱即用的类似mybatis的开发包,找了半天也没有,算了,不如自己造轮子,这个念头一直在我心中萦绕

无奈,python刚学没多久,稍微有些难度,时间也不太充足,直到遇到了mybatis-mapper2sql,才成为可能,感谢大佬

项目地址

特别提醒

暂未发布开发包,只能下载使用

私以为未经过测试就发布,容易挨骂,所以等等看,看看反馈吧

mybatis 语法支持

标签支持

• sql,select,update,insert,delete

• include,if,choose,when,otherwise

• trim,where,set,foreach,bind

动态语法支持

• #{},${}
• 新增 $f{}函数语法 ,允许开发者自行注册函数,参见:mapper_func.py
• 废弃ognl语法 改为 python语法
• 参数转换(核心代码在type_handler.py)

sql 参数支持

参数 目前只支持 dict 类型 名称为params

结果映射

暂不支持,查询结果参见pymysql,pymysql.cursors.DictCursor

数据库支持

• mysql
  理论上可支持所有sql类型数据库

使用示例

基本示例

测试表结构

DROP TABLE IF EXISTS `fruits`;CREATE TABLE `fruits` (`id` bigint(20) NOT NULL,`name` varchar(20) DEFAULT NULL,`category` varchar(20) DEFAULT NULL,`price` double DEFAULT NULL,`create_time` datetime DEFAULT NULL,`type` int(11) DEFAULT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;insert  into `fruits`(`id`,`name`,`category`,`price`,`create_time`,`type`) values (1,'苹果','apple',5,'2020-06-27 11:54:38',0),(2,'梨','pear',4,'2020-06-28 11:54:38',1),(3,'香蕉','banana',3,'2020-06-30 11:55:19',2);

PdbcSqlTemplate(sql模板用法)

参见:tests/sql_template_test.py

from py_mybatis.sql.pdbc_sql_template import *
import pymysqlimport unittestdef query_function(connection):cursor = connection.cursor()try:cursor.execute('select * from fruits')data = cursor.fetchall()connection.commit()return datafinally:cursor.close()class PyMybatisTest(unittest.TestCase):@classmethoddef setUpClass(cls):cls.sql_template = PdbcSqlTemplate(dataSource=PooledDB(creator=pymysql,maxconnections=6,mincached=2,maxcached=5,blocking=True,maxusage=None,setsession=[],ping=0,host="localhost",user="root",password="root",database="api_user",cursorclass=pymysql.cursors.DictCursor,charset='utf8'))def test_select_list(self):print("============{}============".format('select_with row_bound'))print(self.sql_template.select_list(sql="select * from fruits where id in (%s,%s,%s)",row_bound=RowBound(1, 2), args=(1, 2, 3)))print("============{}============".format('select_only'))print(self.sql_template.select_list(sql="select * from fruits where id in (%s,%s,%s)", args=(1, 2, 3)))def test_select_one(self):print("============{}============".format('test_select_one'))print(self.sql_template.select_one(sql="select * from fruits where id=%s", args=(1)))def test_select_with_no_params(self):print("============{}============".format('test_select_with_no_params'))print(self.sql_template.select_one(sql="select * from fruits where id=1"))def test_select_page(self):print("============{}============".format('test_select_page'))print(self.sql_template.select_page(sql="select * from fruits", row_bound=RowBound(1, 2)))def test_delete(self):print("============{}============".format('test_delete'))row = self.sql_template.delete(sql="delete from fruits where id in(%s,%s,%s)",args=(10, 11, 9))print(row)def test_insert_one(self):print("============{}============".format('test_insert_one'))row = self.sql_template.insert(sql="insert into fruits(id,name,price,category) values(%s,%s,%s,%s)",args=(10, '菠萝', 10, 'boluo'))print(row)def test_insert_mul(self):print("============{}============".format('test_insert_mul'))row = self.sql_template.insert_batch(sql="insert into fruits(id,name,price,category) values(%s,%s,%s,%s)",args=((11, '菠萝', 11, 'boluo11'), (9, '菠萝1', 9, 'boluo9')))print(row)def test_query_in_connection(self):print("============{}============".format('test_query_in_connection'))print(self.sql_template.execute_in_connection(query_function))if __name__ == '__main__':unittest.main()

MybatisSqlSession 使用

• mapper.xml

  <?xml version="1.0" encoding="UTF-8"?>
  <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
  <mapper namespace="Test"><sql id="sometable">fruits</sql><sql id="somewhere">WHEREcategory = #{category}</sql><sql id="someinclude">FROM<include refid="${include_target}"/><include refid="somewhere"/></sql><delete id="deleteById">delete from fruits where id =#{id}</delete><select id="testBasic">SELECTid,name,category,priceFROMfruitsWHEREcategory = 'apple' AND<![CDATA[ price < 500 ]]></select><select id="test_function">SELECTname,category,priceFROMfruitsWHEREcategory =$f{like(params['name'])}and create_time =$f{time_format(params['create_time'])}</select><select id="testParameters">SELECTid,name,category,priceFROMfruitsWHEREcategory = #{category,sql_type=VARCHAR}AND price > ${price,sql_type=BIGINT}AND type = ${type,sql_type=BOOLEAN}</select><select id="testInclude">SELECTname,category,price<include refid="someinclude"><property name="prefix" value="Some"/><property name="include_target" value="sometable"/></include></select><select id="testIf">SELECTname,category,priceFROMfruitsWHERE1=1<if test="'category' in params">AND category = #{category}</if><if test="'price' in params">AND price = ${price}<if test="params['price'] >= 400">AND name = 'pear'</if></if></select><select id="testTrim">SELECTname,category,priceFROMfruits<trim prefix="WHERE" prefixOverrides="OR">OR category = 'apple'OR price = 200</trim><trim prefix="AND" prefixOverrides="OR">(type = 1 OR type= 0)</trim></select><select id="testWhere">SELECTname,category,priceFROMfruits<where>AND category = 'apple'<if test="'price' in params ">AND price = ${price}</if></where></select><update id="testSet">UPDATEfruits<set><if test="'category' in params">category = #{category},</if><if test="'price' in params">price = ${price}</if></set>WHEREname = #{name}</update><select id="testChoose">SELECTname,category,priceFROMfruits<where><choose><when test="'name' in params">AND name = #{name}</when><when test="'category' in params and  params['category'] == 'banana'">AND category = #{category}<if test="'price' in params">AND price = ${price}</if></when><otherwise>AND category = 'apple'</otherwise></choose></where></select><select id="testForeach">SELECTname,category,priceFROMfruits<where>category = 'apple' AND name in<foreach collection="names" item="name" open="(" close=")" separator=",">#{name}</foreach></where></select><insert id="testInsertMulti">INSERT INTOfruits(id,name,category,price,create_time)VALUES<foreach collection="fruits" item="fruit" separator=",">(#{fruit['id']},#{fruit['name']},#{fruit['category']},#{fruit['price']},$f{time_format(fruit['create_time'])})</foreach></insert><select id="testBind"><bind name="likeName" value="'%{}%'.format(params['name'])"/>SELECTname,category,priceFROMfruitsWHEREname like #{likeName}</select><insert id="testInsertSelective">insert into fruits<trim prefix="(" suffix=")" suffixOverrides=","><if test="'name' in params">name,</if><if test="'category' in params">category,</if><if test="'price' in params">price,</if></trim><trim prefix="values (" suffix=")" suffixOverrides=","><if test="'name' in params">#{name},</if><if test="'category' in params">#{category},</if><if test="'category' in params">#{price},</if></trim></insert><select id="testChooseNative">SELECTname,category,priceFROMfruits<where><choose><when test="'name' in params">AND name = #{name}</when><when test="'category' in params and params['category'] == 'banana'">AND category = #{category}<if test="'price' in params">AND price = ${price}</if></when><otherwise>AND category = 'apple'</otherwise></choose></where></select><insert id="insertOne">insert into fruits(id,name,category,price) values(#{id},#{name},#{category},#{price})</insert>
  </mapper>

• 示例代码

from py_mybatis.sql.mybatis_sql_session import MybatisMapperScanner, MybatisSqlSession, PooledDB
import pymysql
import osimport unittestimport timeclass PyMybatisTest(unittest.TestCase):@classmethoddef setUpClass(cls):path = os.path.abspath("./")mapper_scanner = MybatisMapperScanner()mybatis_mapper_dict = mapper_scanner.mapper_xml_scan(mapper_xml_dir=path)pool = PooledDB(creator=pymysql,maxconnections=6,mincached=2,maxcached=5,blocking=True,maxusage=None,setsession=[],ping=0,host="localhost",user="root",password="root",database="api_user",cursorclass=pymysql.cursors.DictCursor,charset='utf8')cls.sql_session = MybatisSqlSession(mapper_dict=mybatis_mapper_dict, dataSource=pool)cls.sql_namespace = 'Test.'def test_base(self):self.sql_id = self.sql_namespace + 'testBasic'print("============{}============,sql_result:{}".format(self.sql_id, self.sql_session.select_list(self.sql_id)))def test_parameters(self):self.sql_id = self.sql_namespace + 'testParameters'params = {'category': 'apple', 'price': '15.0', 'type': '001', 'content': '图书'}print("============{}============,sql_result:{}".format(self.sql_id, self.sql_session.select_list(self.sql_id,params=params)))def test_include(self):self.sql_id = 'testInclude'params = {'category': 'apple'}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_list(self.sql_id,params=params)))def test_if(self):self.sql_id = 'testIf'params = {'category': 'apple', 'price': 5}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_one(self.sql_id,params=params)))def test_trim(self):self.sql_id = 'testTrim'self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_one(self.sql_id)))def test_where(self):self.sql_id = 'testWhere'params = {'category': 'apple', 'price': 500}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_one(self.sql_id,params=params)))def test_set(self):self.sql_id = 'testSet'params = {'category': 'apple', 'price': 500, 'name': 'name1'}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.update(self.sql_id,params=params)))def test_choose(self):self.sql_id = 'testChoose'params = {'age': 1, 'sex': '1', 'name': 'name', 'price': 'price', 'category': 'banana'}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_list(self.sql_id,params=params)))def test_foreach(self):self.sql_id = 'testForeach'params = {'names': ['name1', 'name2', 'name3']}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_list(self.sql_id,params=params)))def test_bind(self):self.sql_id = 'testBind'params = {'name': 'name'}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_list(self.sql_id,params=params)))def test_choose_native(self):self.sql_id = 'testChooseNative'params = {'category': 'banana', 'price': 500, 'name': 'name1'}self.sql_id = self.sql_namespace + self.sql_idprint("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.select_list(self.sql_id,params=params)))def test_insert_multi(self):self.sql_id = 'testInsertMulti'self.sql_id = self.sql_namespace + self.sql_idcreate_time = time.localtime()params = {'fruits': [{'id': 1,'name': 'apple','category': 'apple','price': 5.0,'create_time': create_time},{'id': 2,'name': 'banana','category': 'banana','price': 3.0,'create_time': create_time},{'id': 3,'name': 'pear','category': 'pear','price': 4.0,'create_time': create_time},]}print("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.insert(self.sql_id,params=params)))def test_insert_selective(self):self.sql_id = 'testInsertSelective'self.sql_id = self.sql_namespace + self.sql_idparams = {'name': 'apple','category': 'apple','price': 5.0}print("============{}============,sql_result:{}".format(self.sql_id,self.sql_session.insert(self.sql_id,params=params)))"""self.sql_session.begin_tx():会在当前线程的thread_local下绑定一个connection 此后执行所有方法都会在一个 connection下除非手动调用 self.sql_session.close(): 关闭连接self.sql_session.commit():提交事务self.sql_session.rollback(): 关闭事务未调用:self.sql_session.begin_tx()方法:每次调用sql_session 都会新建一个连接,建议使用连接池参考用法:try:self.sql_session.begin_tx().... do_something_txself.sql_session.commit()print("commit")except Exception as e:print("exception rollback", e)self.sql_session.rollback()finally:print("finally close")self.sql_session.close()"""def test_tx(self):print("============{}============".format('test_tx'))"""template method"""fruits = [{'id': 4,'name': 'apple','category': 'apple','price': 5.0},{'id': 5,'name': 'banana','category': 'banana','price': 3.0},{'id': 6,'name': 'pear','category': 'pear','price': 4.0},]try:self.sql_session.begin_tx()for index, item in enumerate(fruits):self.sql_session.insert(sql_id='Test.insertOne', params=item)if index > 1:raise Exception('rollback test')self.sql_session.commit()print("commit")except Exception as e:print("exception rollback", e)self.sql_session.rollback()finally:print("finally close")self.sql_session.close()if __name__ == '__main__':unittest.main()