代码收藏家技术教程 2024-12-31

Python版的Mybatis ORM：mybatis-py（类似java的mybatis）

入门

mybatis-py是一个和mybatis很相似的轻量级的半自动的ORM，它的功能如下：

主要功能如下：
1 、半自动化的 ORM ；
2 、支持像 mybatis 那样的动态 SQL ；
3 、支持类似 mybatis 注解的装饰器；
4 、支持 LRU 缓存，支持缓存过期机制；
5 、尽可能地使用 Prepared statement ，避免 SQL 字符串拼接，能有效地防止 SQL 注入；
6 、预防大对象机制，避免 OOM 。

7、支持mysql和postgresql

它的地址在https://github.com/ralgond/mybatis-py

首先我们先来安装它：

pip install -U mybatis

接着创建数据库，用户名、密码、数据库名均为mybatis：

CREATE DATABASE mybatis;

USE mybatis;

CREATE TABLE IF NOT EXISTS fruits (
        id INT AUTO_INCREMENT PRIMARY KEY, 
        name VARCHAR(100),
        category VARCHAR(100),
        price int)

INSERT INTO fruits (name, category, price) VALUES ('Alice', 'A', 100)
INSERT INTO fruits (name, category, price) VALUES ('Bob', 'B', 200)

接下来就可以编写代码了，首先创建一个python文件test.py，内容如下：

from mybatis import Mybatis, ConnectionFactory

conn = ConnectionFactory.get_connection(
    dbms_name='mysql', # change to 'postgresql' if you are using PostgreSQL
    host="localhost",
    user="mybatis",
    password="mybatis",
    database="mybatis")

mb = Mybatis(conn, "mapper", cache_memory_limit=50*1024*1024)

@mb.SelectOne("SELECT * FROM fruits WHERE id=#{id}")
def get_one(id:int):
    pass

@mb.SelectMany("SELECT * FROM fruits")
def get_many():
    pass

@mb.Insert("INSERT INTO fruits (name, category, price) VALUES (#{name}, #{category}, #{price})", primary_key="id")
def insert(name:str, category:str, price:int):
    pass

@mb.Delete("DELETE FROM fruits WHERE id=#{id}")
def delete(id:int):
    pass

@mb.Update("UPDATE fruits SET name=#{name} WHERE id=#{id}")
def update(name:str, id:int):
    pass

print(get_one(id=1))

print(delete(id=4))

print(get_many())

print(insert(name="Dating", category="D", price=20))

print(get_many())

print(update(name='Amazon', id=1))

print(get_many())

以上是利用装饰器来实现的，是不是很简单？

但是如果您的SQL比较复杂，我推荐使用xml mapper文件的方式，如下：

首先创建一个mapper/test.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>
    <insert id="testInsert1">
        INSERT INTO fruits (name, category, price) VALUES (#{name}, #{category}, #{price})
        <if test="'__need_returning_id__' in params">
            RETURNING ${__need_returning_id__}
        </if>
    </insert>
    <select id="testBasic1">
        SELECT * from fruits where id=#{id}
    </select>
</mapper>

再创建一个test2.py，内容是：

from mybatis import *

def main():
    conn = ConnectionFactory.get_connection(
            dbms_name='mysql', # change to 'postgresql' if you are using PostgreSQL
            host="localhost",
            user="mybatis",
            password="mybatis",
            database="mybatis")

    mb = Mybatis(conn, "mapper", cache_memory_limit=50*1024*1024)

    ret = mb.insert('testInsert1', {'name':'Alice', 'category':'C', 'price':500})
    ret = mb.select_one("testBasic1", {'id':1})

    print(ret)

if __name__ == "__main__":
    main()

这样就OK啦。

缓存

mybatis-py和mybatis一样，也支持缓存，它会计算缓存对象的占用的字节数，再加入新对象时判断容量是否已经不够了，如果不够，就采用LRU算法来淘汰一些对象。

另外缓存是由时间限制的，用户可以在创建Mybatis对象（每个对象都有一个缓存池）时指定超时时间。

安全

mybatis-py在能使用prepared statement的地方都尽可能使用了，以减少SQL注入的风险。

避免OOM

为了避免从数据库拉出来的数据过大，可以设置返回对象的总字节数上限，这样能有效地降低OOM的可能性。

实战

下面是一段实战代码，基于Flask框架

from flask import Flask

import mybatis.errors
from mybatis import Mybatis, ConnectionFactory
import orjson as json
import functools

app = Flask(__name__)


# 连接到 MySQL 数据库
conn = None
mb = Mybatis(conn, "mapper", cache_memory_limit=50*1024*1024)
connection_error = False
error_string = ""

def make_connection_and_mybatis():
    global conn
    global mb
    global connection_error
    global error_string

    if conn is None:
        try:
            conn = ConnectionFactory.get_connection(
                            dbms_name="postgresql",
                            host="localhost",
                            user="mybatis",  
                            password="mybatis", 
                            database="mybatis"
            )
            mb.conn = conn
            mb.conn.set_autocommit(False)
                
            return True
            
        except Exception as e:
            connection_error = True
            error_string = str(e)
            return False
    else:
        try:
            if connection_error:
                conn.reconnect(3, 3)
                connection_error = False

            if mb.conn is None:
                mb.conn = conn
            return True
        except Exception as e:
            connection_error = True
            error_string = str(e)
            return False


@mb.SelectOne("SELECT * FROM fruits WHERE id=#{id}")
def select_one(id:int):
    pass

@mb.SelectMany("SELECT * FROM fruits")
def select_many():
    pass

@mb.Insert("INSERT INTO fruits (name,category,price) VALUES ('Candy', 'C', 500)")
def insert():
    pass

def sql_auto_reconnect(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        global connection_error
        try:
            ret = make_connection_and_mybatis()
            if ret is False:
                return error_string, 500

            ret = func(*args, **kwargs)
            return ret, 200

        except mybatis.errors.DatabaseError as e:
            connection_error = True
            return str(e), 500
        except Exception as e:
            return str(e), 500

    return wrapper


@app.route('/')
@sql_auto_reconnect
def hello():
    ret = select_many()
    return json.dumps(ret)


@app.route('/insert')
@sql_auto_reconnect
def do_insert():
    ret = insert()
    return json.dumps(ret)

if __name__ == "__main__":
    app.run(debug=True)

作者：moontourse