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

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