python - 是否可以在 motor 中对集合进行子类化？

Question

所以我正在使用Motor在我的应用程序中，我希望创建用于与数据库交互的辅助函数，我最初考虑使用 MongoTurbine但我偏向于使用 ORM，尽管让例程函数更容易调用会很好。

让我们以更新插入为例:

我可以使用以下内容编写更新插入:

await collection.update_one({'key': value}, {'set': {'key': value}}, upsert = True)

但是使用起来会更容易:

await collection.upsert({'key': value}, {'set': {'key': value}})

或者甚至是一个set方法:

await collection.set({'key': value}, {'key': value})

我有很多类似的方法，还有一些更复杂的方法，但如果有人能指出我正确的方向，那就太棒了!

促使我问这个问题的触发因素是我看到 AsyncIOMotorDatabase 中有一个 document_class 参数。它允许您指定文档的返回类型。但没有真正简单的方法来指定自定义集合类。

我知道所有这些都可以使用标准约定来完成，我只是想让事情变得更容易，因为有时可能会有一些很长的过滤器、更新、投影字典，我只是想实现我自己的约定。

编辑

基本上，为了将更多上下文放入其中，我希望创建自定义函数来减少我需要一遍又一遍地编写的字典数量。以更新用户的积分为例；我希望能够围绕 update_one 函数编写一种“包装”函数，以提高跨多个模块的可读性和可用性。

例如:

async def update_set(key, value, **kwargs):
    self.update_one({key: value}, {'$set': kwargs})

await db.users.update_set('username', 'johndoe', foo = 1, bar = 'test')
#vs
await db.users.update_one({'username': 'johndoe'}, {'$set': {'foo': 1, 'bar': 'batz'}})

Answer 1

我尝试了简单的子类化，但它不起作用:

class MyCollection(motor_asyncio.AsyncIOMotorCollection):
    pass

my_collection = MyCollection(database=db, name='my_collection')

Traceback (most recent call last):
  File "./test.py", line 13, in <module>
    my_collection = MyCollection(database=db, name='my_collection')
  File "/usr/local/lib/python3.6/dist-packages/motor/core.py", line 528, in __init__
    super(self.__class__, self).__init__(delegate)
TypeError: __init__() missing 1 required positional argument: 'name'

但是通过一些技巧，它终于起作用了:

#!/usr/bin/python3.6

import pymongo
from motor import motor_asyncio


class MyCollection(motor_asyncio.AsyncIOMotorCollection):
    def __init__(self, *args, **kwargs):
        """Calling __init__ of parent class is failing for some reason"""

    def __new__(cls, *args, **kwargs):
        collection = motor_asyncio.AsyncIOMotorCollection(*args, **kwargs)
        collection.__class__ = MyCollection
        return collection

    def my_method(self, *args, **kwargs):
        print('my method', args, kwargs)


client = motor_asyncio.AsyncIOMotorClient()
db = client.db
my_collection = MyCollection(database=db, name='my_collection')
my_collection.my_method()  # my_method () {}
assert db.my_collection == my_collection  # passes assertion