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

平凡888 / py_mybatis

py_mybatis

项目起源

mybatis 语法支持

标签支持

动态语法支持

sql 参数支持

结果映射

数据库支持

使用示例

基本示例

PdbcSqlTemplate(sql模板用法)

MybatisSqlSession 使用

简介

发行版

贡献者

近期动态

平凡888 / py_mybatis .gitee-modal { width: 500px !important; }

py_mybatis

项目起源

mybatis 语法支持

标签支持

动态语法支持

sql 参数支持

结果映射

数据库支持

使用示例

基本示例

PdbcSqlTemplate(sql模板用法)

MybatisSqlSession 使用

简介

发行版

贡献者

近期动态

搜索帮助

平凡888 / py_mybatis