# easemysql

**Repository Path**: jaifire/easemysql

## Basic Information

- **Project Name**: easemysql
- **Description**: 简单易用的 mysql 封装
- **Primary Language**: Unknown
- **License**: MIT
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-03-19
- **Last Updated**: 2026-05-12

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# EaseMySQL

基于 `pymysql` + `DBUtils` 连接池的 MySQL 封装库，提供简洁的 CRUD 接口、灵活的条件构造器和内存查询能力。

## 安装

```bash
pip install --upgrade easemysql -i https://pypi.org/simple/
```

依赖：

```bash
pip install pymysql --upgrade
pip install DBUtils --upgrade
```

## 快速开始

```python
from easemysql import EaseMySQL

db = EaseMySQL(host='localhost', user='root', password='123456', database='mydb')

# 查询单条
row = db.one('users', dict(id=1))

# 查询多条
rows = db.all('users', 'status = 1')

# 插入
db.insert('users', dict(name='test', age=20))

# 更新
db.update('users', dict(age=21), 'id = 1')

# 删除
db.delete('users', 'id = 1')
```

## EaseMySQL 接口

### 查询

```python
# 单条记录，返回 dict 或 None
db.one(table, condition, params=(), field='*', order=None, group=None, flat=False)

# 多条记录，返回 list[dict]
db.all(table, condition='1', params=(), field='*', group=None, order=None, limit=None, list_query=False)
```

- `flat=True` 时 `one` 直接返回第一个字段的值（如只查 `id`）
- `list_query=True` 时 `all` 返回 `ListQuery` 对象（支持内存二次查询，见下文）
- `field` 指定查询字段，如 `'id, name'`

### 插入

```python
# 单条插入，默认 INSERT
db.insert('users', dict(name='test', age=20))          # INSERT INTO
db.insert('users', dict(name='test', age=20), mode='ignore')   # INSERT IGNORE INTO
db.insert('users', dict(name='test', age=20), mode='replace')  # REPLACE INTO

# 批量插入，默认 REPLACE
db.insert('users', [dict(name='a'), dict(name='b')])           # REPLACE INTO
db.insert('users', [dict(name='a'), dict(name='b')], mode='insert')  # INSERT INTO
```

> **注意**：`REPLACE` 会覆盖同主键行，未传入的字段会被置为默认值。

### 更新

```python
db.update(table, data, condition, params=())

# 批量更新（参数化）
# data: [{field: value}, ...]  params: [[where_val], ...]
db.update_many('users',
    data=[{'name': 'A'}, {'name': 'B'}],
    condition='id = %s',
    params=[[1], [2]]
)
```

### 删除

```python
db.delete(table, condition, params=())
```

### 原生 SQL

```python
# 执行写操作，返回 (lastrowid, rowcount)
db.raw('UPDATE users SET age = %s WHERE id = %s', [25, 1])

# 查询单条
db.raw_one('SELECT * FROM users WHERE id = %s', [1], flat=False)

# 查询多条，list_query=True 返回 ListQuery
db.raw_all('SELECT * FROM users WHERE age > %s', [18])
```

### 事务

```python
with db.transaction():
    db.update('accounts', dict(balance=100), 'id = 1')
    db.update('accounts', dict(balance=200), 'id = 2')
```

事务支持嵌套（内部嵌套不会产生新事务）。

### update_or_create

```python
# 按条件查找，存在则更新 defaults，不存在则创建
instance, created = db.update_or_create(
    'users',
    defaults=dict(age=25),
    create_defaults=dict(created_at='2024-01-01'),
    name='test'
)
```

### JOIN 查询

```python
# table 参数传入 [主表, [join列表]]
rows = db.all(
    ['orders o', [
        ['users u', 'o.user_id = u.id'],                          # 默认 LEFT JOIN
        ['products p', 'o.product_id = p.id', 'INNER JOIN'],     # 指定 JOIN 类型
    ]],
    condition='o.id > 0',
    field='o.id, u.name, p.title'
)
```

### last_sql

每次操作后可通过 `db.last_sql` 获取最后执行的 SQL 语句。

## 条件构造

`where_simple` / `where_complex` 支持三种条件类型：字符串、字典、列表。

### 字符串

```python
db.all('users', 'age > 18')
db.all('users', 'age > %s', [18])  # 参数化
```

### 字典（AND 连接）

```python
# 值为列表时自动转为 IN
db.one('users', dict(id=1))
db.all('users', dict(status=1, type=[2, 3]))
# → status = 1 AND type IN (2,3)
```

### 列表（灵活组合）

```python
# AND（默认）
db.all('users', ['age > 18', 'status = 1'])
# → age > 18 AND status = 1

# OR
db.all('users', ['OR', dict(status=1), dict(role='admin')])
# → status = 1 OR role = 'admin'

# 比较运算符
db.all('users', ['>=', 'age', 18])
# → age >= 18

# IN / NOT IN
db.all('users', ['IN', 'id', [1, 2, 3]])
db.all('users', ['NOT IN', 'id', [4, 5]])

# BETWEEN
db.all('users', ['BETWEEN', 'age', 18, 30])

# NOT
db.all('users', ['NOT', dict(status=0)])

# 嵌套组合
db.all('users', [
    'OR',
    ['AND', dict(status=1), 'age >= 18'],
    ['IN', 'role', ['admin', 'editor']],
    dict(id=100),
])
```

### where_in

```python
from easemysql import where_in

where_in([1, 2, 3])       # → '1,2,3'
where_in(['a', 'b'])      # → "'a','b'"
where_in([])               # → 'NULL'（空列表返回 NULL，IN (NULL) 恒为 FALSE）
where_in('1,2,3')          # → '1,2,3'（字符串原样返回）
```

## ListQuery 内存查询

`db.all(..., list_query=True)` 返回 `ListQuery` 对象，支持对查询结果进行内存二次筛选和聚合。

```python
from easemysql import EaseMySQL

db = EaseMySQL(host='localhost', user='root', password='123456', database='mydb')
lq = db.all('orders', dict(status=1), list_query=True)
```

### filter

```python
lq.filter(status=1)
lq.filter(age__gt=18)       # 大于
lq.filter(age__gte=18)      # 大于等于
lq.filter(age__lt=60)       # 小于
lq.filter(age__lte=60)      # 小于等于
lq.filter(age__ne=0)        # 不等于
lq.filter(name__contains='张')  # 包含
lq.filter(id__in=[1,2,3])   # 在列表中

# OR 条件（传多个字典）
lq.filter({'role': 'admin'}, {'role': 'editor'})

# 链式调用
lq.filter(status=1).filter(age__gt=18)
```

### order_by

```python
lq.order_by('age')       # 升序
lq.order_by('-age')      # 降序
lq.order_by('status', '-age')  # 多字段排序
```

### group_by

```python
lq.group_by('dept', sum='salary', count='id')
# → ListQuery，每项包含 {dept, sum(salary), count(id)}
```

### 聚合

```python
lq.sum('salary')                    # 求和，默认 int 转换
lq.sum('salary', pre=float)         # 自定义转换
lq.avg('salary')                    # 平均值
lq.avg('salary', pre=float, fmt=lambda x: round(x, 2))  # 自定义精度
lq.count()                          # 记录数
```

### 取值

```python
lq.first()                          # 第一条，无数据返回 None
lq.get()                            # 返回原始 list[dict]
lq.get_map('id')                    # → {id: dict}，单字段做 key
lq.get_map('dept', 'team')          # → {'dept@team': dict}，多字段用 @ 拼接做 key
lq.values_list('id')                # → [1, 2, 3]（flat 模式，单字段）
lq.values_list('id,name', flat=False)  # → [[1, 'a'], [2, 'b']]
len(lq)                             # 等同于 lq.count()
for item in lq:                     # 可迭代
    print(item)
```


## 连接池

`SQLSession` 使用 `@singleton` 装饰器，相同连接参数只会创建一个连接池实例：

```python
from easemysql import EaseMySQL

# 以下两个 db 共享同一个连接池（参数相同）
db1 = EaseMySQL(host='localhost', user='root', password='123456', database='mydb')
db2 = EaseMySQL(host='localhost', user='root', password='123456', database='mydb')
```

连接池参数（通过 EaseMySQL 构造函数传入）：

| 参数 | 默认值 | 说明 |
|------|--------|------|
| `maxconnections` | 10 | 最大连接数 |
| `mincached` | 1 | 初始连接数 |
| `maxcached` | 10 | 最大空闲连接数 |
| `blocking` | True | 连接用尽时是否等待 |

## 注意事项

- **LIKE 中的 `%`**：pymysql 将 `%` 视为格式化占位符，不能直接执行 `LIKE '%text%'`，需使用参数化查询：
  ```python
  db.raw_all("SELECT * FROM users WHERE name LIKE %s", ["%test%"])
  ```
- **REPLACE 行为**：批量插入默认使用 `REPLACE INTO`，会覆盖同主键行并将未传入字段置为默认值。