这段时间写的东西由于场景需要,从之前的mysql全面迁移到了mongo,于是开始寻找有没有能简化开发流程的东西比如orm,对比几种社区给出的建议,最终选择了mongoengine,这篇博客用来记录一下开发中遇到的一些问题与解决方案

mongoengine支持的MongoDB版本是 3.4, 3.6, 4.0,版本不对的话使用过程中会遇到 FieldPath connot be constructed with empty string 的异常

查询条件为ObjectId

众所周知mongo中的 _id 并不是个字符串,所以要查询之前得先转格式,MongoEngine的QuerySet提供了一个比较方便的with_id

1
2
3
4
- from bson.objectid import ObjectId

- TestModel.objects.get(id=ObjectId('xxxxx'))
+ TestModel.objects.with_id('xxxxx')

字段格式校验

关于这个其实感觉设计有点蠢,看完实现代码发现如果有一个需求是在保存的时候,某个文档中一些字段需要验证,而一些字段不需要验证,靠原生方法是实现不了的

Document 的保存函数中,有一个 validate 的设置项,如果存在的话会调用 BaseDocument 里的 validate(self, clean=True),这个函数会把该文档下所有字段的 _validate 函数调用一下。这里就出现了一个让我有点难以理解的设计,不管这个字段设没设置 validation=None,它最终都要调用一下 self.validate(value, **kwargs),每个继承自 BaseField 的字段,StringField, DatetimeField… 都有实现这个 validate 函数,而这个函数里他妈根本就没去判断这个字段有没有 validation,直接就开始做校验了,

所以想要实现字段不校验,要么整个文档都不校验,要么整个文档都校验,除非自己再来继承一下这个字段类型,把 validate 给重写了

1
2
3
4
5
6
7
8
9
from mongoengine.fields import DateTimeField

class CustomDatetimeField(DateTimeField):
def validate(self, value):
if self.validation is not None:
super().validate(value)

class TestModel(Document):
published_at = CustomDatetimeField(validation=None)

ListField 模糊查询

如果在定义了一个字段是 ListField, 这时候有一个需求是如果我想搜索tags中存在bc的值

1
2
3
4
class TestModel(Document):
tags = ListField()

TestModel(tags=['abc', 'bcd', 'ccc']).save()

常见ListField中的搜索一般为 TestModel.objects(tags="abc"),而这只是完全匹配,如果条件为 tags="cd" 那么返回将会是0条,实现模糊匹配的方式可以使用正则

1
2
3
import re

TestModel(tags=re.compile('.*%s.*', re.IGNORECASE) % search)

实现简单的lucene语法查询

使用 luqum 来做解析, pip install luqum

不考虑更复杂的情况,满足简单需求,如

1
2
3
title: "xxx"
name: "aaa" AND title: "xxx"
(name: "aaa" AND title: "xxx") OR socre: 100

MongoEngine 中这种复杂查询,可以使用 Q 来满足

1
2
3
from mongoengine.queryset.visitor import Q

TestModel.objects.filter((Q(name__icontains="xxx") & Q(title__icontains="xxx")))

这三个查询例子中,说到底其实需要5种类型,SearchField, OrOperation, AndOperation, Group, Phrase

所以在解析查询语句后,需要来判断一下类型

1
2
3
4
5
6
7
8
9
10
from luqum.parser import parser

tree = parser.parse(search_text)
def search_parser(tree):
if isinstance(tree, SearchField):
pass
elif isinstance(tree, (OrOperation, AndOperation)):
pass
elif isinstance(tree, Group):
pass

而最终所有的类型都会走向 SearchField,那么只需要再写一个解析字段的函数,返回一个 Q,而这里有个需要注意的,Q 里的值不能写成动态的,比如

1
2
3
?search=abc
search = request.args.get('search')
Q(search="xx")

结果将会是查的search而不是abc,所以继承一下Q来写一个自定义的CQ

1
2
3
class CQ(Q):
def __init__(self, name, value):
self.query = {'%s__icontains' % name.lower(): value}

再使用CQ来进行字段的处理

1
2
3
4
5
6
7
8
from mongoengine.queryset.visitor import Q

def parse_search_field(node: SearchField) -> Q:
name = node.name
value = node.expr.__str__(head_tail=False)
if isinstance(node.expr, Phrase):
value = value.lstrip('"').rstrip('"')
return CQ(name, value)

这里也可以自定义一个 ACCEPT_LIST 来筛选可搜索的字段,然后来处理 Group,这个比较简单,就是一个只有一个值的列表,所以直接返回就好了

1
2
if isinstance(tree, Group):
return (search_parser(tree.children[0]))

最后再来处理 and 和 or 的情况,跟完 MongoEngine 中关于这块处理的代码发现,最终使用的是 QCombination 这个类来做的组合

1
2
3
4
5
6
if isinstance(tree, (OrOperation, AndOperation)):
nodes = []
op = 0 if tree.op == 'AND' else 1
for item in tree.children:
nodes.append(search_parser(item))
return QCombination(op, nodes)

最后直接调用 search_parser 即可

1
2
3
search = 'name: "aaa" AND title: "xxx") OR socre: 100'
_filter = search_parser(parser.parse(search))
TestModel.objects.filter(_filter)