记如何在 Tortoise Orm 中使用 Mysql 之 Json 查询

发表于 更新于 分类于 阅读次数： 本文字数： 265 阅读时长 ≈ 1 分钟

前记

在设计数据库的时候有一个 IP 字段，是用来存多个 IP 地址。于是设计成了 Json 类型，记一下如何在 Tortoise Orm 中使用 JSON_CONTAINS

使用

Model

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from tortoise import Model, fields
from uuid import uuid4

class ag_ip_group(Model):
    id = fields.IntField(pk=True)
    uuid = fields.UUIDField(default=uuid4)
    name = fields.CharField(64)
    ip = fields.JSONField()
    remark = fields.CharField(64,default='_')
    status = fields.IntField(default=1)
    created_time = fields.DatetimeField(null=True, auto_now_add=True)
    updated_time = fields.DatetimeField(null=True, auto_now=True)

    def __str__(self):
        return str(self.id)

Function

1
2
3
4
5
from tortoise.functions import Function
from pypika import CustomFunction

class JsonContains(Function):
    database_func = CustomFunction("JSON_CONTAINS", ["field", "value"])

Service

1
2
3
4
5
6
filter = {}
annotate = {}
annotate['json_len'] = JsonContains('ip', '"'+row['ip']+'"')
filter['json_len__gt'] = 0

await self.model.all().annotate(**annotate).filter(**filter).limit(limit).offset(offset).values()

扩展

其他的 Json 函数也可自行扩展，这里我们再扩展一种 Json 模糊查询，即：

1
select * from table where json_extract(field, '$') LIKE '%value%'

实现如下：

1
2
3
4
5
6
7
8
9
class JsonExtract(Function):
    database_func = CustomFunction("JSON_EXTRACT", ["field", 'value'])

filter = {}
annotate = {}
annotate['json_str'] = JsonExtract('ip','$')
filter['json_str__icontains'] = row['ip']

await self.model.all().annotate(**annotate).filter(**filter).limit(limit).offset(offset).values()

其他

在 JSONField 类型字段中使用 encoder 更改 json.dumps() 参数 ensure_ascii=False，实现也非常的简单

1
2
import json
field = fields.JSONField(encoder=lambda x: json.dumps(x,ensure_ascii=False))

锦城虽云乐，不如早还家