- python的mybatis实现
- python的SqlTemplate实现
特别鸣谢:mybatis-mapper2sql
借助这位大佬使我快速实现了py_mybatis,mapper xml解析主体代码都是 hhyo大佬的
本人是位javaer,刚刚开始学习python,当开始学习数据库相关api时,发现python中并未有很好的开发包,
重型orm 直接淘汰,不要问为什么,问就是我喜欢写sql,虽然sql写的不咋的
急需一种轻型的,开箱即用的类似mybatis的开发包,找了半天也没有,算了,不如自己造轮子,这个念头一直在我心中萦绕
无奈,python刚学没多久,稍微有些难度,时间也不太充足,直到遇到了mybatis-mapper2sql,才成为可能,感谢大佬
特别提醒
暂未发布开发包,只能下载使用
私以为未经过测试就发布,容易挨骂,所以等等看,看看反馈吧
-
sql,select,update,insert,delete
-
include,if,choose,when,otherwise
-
trim,where,set,foreach,bind
- #{},${}
- 新增 $f{}函数语法 ,允许开发者自行注册函数,参见:mapper_func.py
- 废弃ognl语法 改为 python语法
- 参数转换(核心代码在type_handler.py)
参数 目前只支持 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);
参见:tests/sql_template_test.py
from py_mybatis.sql.pdbc_sql_template import *
import pymysql
import unittest
def query_function(connection):
cursor = connection.cursor()
try:
cursor.execute('select * from fruits')
data = cursor.fetchall()
connection.commit()
return data
finally:
cursor.close()
class PyMybatisTest(unittest.TestCase):
@classmethod
def 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()-
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"> WHERE category = #{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"> SELECT id, name, category, price FROM fruits WHERE category = 'apple' AND <![CDATA[ price < 500 ]]> </select> <select id="test_function"> SELECT name, category, price FROM fruits WHERE category =$f{like(params['name'])} and create_time =$f{time_format(params['create_time'])} </select> <select id="testParameters"> SELECT id, name, category, price FROM fruits WHERE category = #{category,sql_type=VARCHAR} AND price > ${price,sql_type=BIGINT} AND type = ${type,sql_type=BOOLEAN} </select> <select id="testInclude"> SELECT name, category, price <include refid="someinclude"> <property name="prefix" value="Some"/> <property name="include_target" value="sometable"/> </include> </select> <select id="testIf"> SELECT name, category, price FROM fruits WHERE 1=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"> SELECT name, category, price FROM fruits <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"> SELECT name, category, price FROM fruits <where> AND category = 'apple' <if test="'price' in params "> AND price = ${price} </if> </where> </select> <update id="testSet"> UPDATE fruits <set> <if test="'category' in params"> category = #{category}, </if> <if test="'price' in params"> price = ${price} </if> </set> WHERE name = #{name} </update> <select id="testChoose"> SELECT name, category, price FROM fruits <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"> SELECT name, category, price FROM fruits <where> category = 'apple' AND name in <foreach collection="names" item="name" open="(" close=")" separator=","> #{name} </foreach> </where> </select> <insert id="testInsertMulti"> INSERT INTO fruits ( 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'])"/> SELECT name, category, price FROM fruits WHERE name 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"> SELECT name, category, price FROM fruits <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 os
import unittest
import time
class PyMybatisTest(unittest.TestCase):
@classmethod
def 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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
print("============{}============,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_id
create_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_id
params = {'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_tx
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()
"""
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()