48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
from typing import List, Dict, Any
|
|
|
|
from pymongo.collection import Collection
|
|
|
|
from config import conf
|
|
from core.exceptions import NotFoundException
|
|
from core.utils import get_time
|
|
|
|
|
|
def get_mongo_url(connection_name:str = None):
|
|
return conf.server.is_prod and conf.database.get(connection_name or 'mongo').production.url or conf.database.get(
|
|
connection_name or 'mongo').dev.url
|
|
|
|
|
|
def get_mongo_db(connection_name:str = None):
|
|
return conf.server.is_prod and conf.database.get(connection_name or 'mongo').production.db or conf.database.get(
|
|
connection_name or 'mongo').dev.db
|
|
|
|
|
|
def nested_join(collection: str, from_field, to_field,
|
|
alias, match: dict) -> List[Dict[str, Any]]:
|
|
"""
|
|
Join a list of nested dictionaries into a single dictionary.
|
|
"""
|
|
return [
|
|
{
|
|
"$lookup": {
|
|
"from": collection,
|
|
"localField": from_field,
|
|
"foreignField": to_field,
|
|
"as": alias
|
|
},
|
|
},
|
|
{
|
|
"$match": match if match else {}
|
|
}
|
|
]
|
|
|
|
|
|
async def use_soft_delete(collection: Collection, find_query: dict) -> bool:
|
|
"""
|
|
Soft delete a document by setting a field to true.
|
|
"""
|
|
result = await collection.update_one(find_query, {"$set": {'deleted_at': get_time()}, })
|
|
if not result.modified_count:
|
|
raise NotFoundException(f"No document found with query: {find_query}")
|
|
return True
|