43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from typing import Union
|
|
|
|
from bson import ObjectId
|
|
|
|
|
|
class CRUDBase:
|
|
def __init__(self, coll_name):
|
|
self.coll_name = coll_name
|
|
|
|
async def get(self, db, id: Union[ObjectId, str]) -> dict:
|
|
return (await db[self.coll_name].find_one({'_id': ObjectId(id)})) or dict()
|
|
|
|
async def find_one(self, db, filter=None, *args, **kwargs):
|
|
return (await db[self.coll_name].find_one(filter, *args, **kwargs)) or dict()
|
|
|
|
async def read_have(self, db, v: str, **kwargs):
|
|
where = {'members': v}
|
|
where.update(kwargs)
|
|
cursor = db[self.coll_name].find(where)
|
|
return await cursor.to_list(length=9999)
|
|
|
|
async def find_many(self, db, **kwargs):
|
|
cursor = db[self.coll_name].find(kwargs)
|
|
return await cursor.to_list(length=9999)
|
|
|
|
def find(self, db, *args, **kwargs):
|
|
cursor = db[self.coll_name].find(*args, **kwargs)
|
|
return cursor
|
|
|
|
@staticmethod
|
|
async def to_list(cursor):
|
|
async for doc in cursor:
|
|
yield doc
|
|
|
|
async def delete(self, db, **kwargs):
|
|
return await db[self.coll_name].delete_many(kwargs)
|
|
|
|
async def update_one(self, db, filter, update, upsert=False):
|
|
return await db[self.coll_name].update_one(filter, update, upsert)
|
|
|
|
async def distinct(self, db, key, filter=None):
|
|
return await db[self.coll_name].distinct(key, filter)
|