Commit 8e66afed by Peter Weidenbach

get_all_values function added

parent cab81724
......@@ -2,6 +2,44 @@ import logging
from bson.son import SON
def get_list_of_all_values(collection, object_path, unwind=False, match=None):
"""
Get a list of unique values on a specific object path in a collection.
An Optional search string (match) can be added.
:param collection: mongo collection to look at
:type collection: pymongo.collection
:param object_path: mongo object path
:type object_path: str
:param unwind: if true: handle list entries as single values
:type unwind: bool
:param match: mongo search string
:type match: dict
:return: list
"""
pipeline = []
if match is not None:
pipeline.append({"$match": match})
pipeline.extend([
{"$group": {"_id": object_path}},
{"$sort": SON([("_id", 1)])}
])
if unwind:
old_pipe = pipeline
pipeline = [{"$unwind": object_path}]
pipeline.extend(old_pipe)
result = _get_list_of_aggregate_list(list(collection.aggregate(pipeline)))
logging.debug(result)
return result
def _get_list_of_aggregate_list(ag_list):
result = []
for item in ag_list:
result.append(item['_id'])
return result
def get_objects_and_count_of_occurrence(collection, object_path, unwind=False, match=None):
"""
Get a list of unique values and their occurences on a specific object path in a collection.
......
from common_helper_mongo.aggregate import get_objects_and_count_of_occurrence,\
get_field_sum, get_field_average
get_field_sum, get_field_average, get_list_of_all_values
import unittest
from tests.base_class_database_test import MongoDbTest
class TestAggregate(MongoDbTest):
def test_get_list_of_all_values(self):
self.add_simple_test_data()
result = get_list_of_all_values(self.test_collection, "$test_txt", unwind=False, match=None)
self.assertIsInstance(result, list, "result not a list")
self.assertEqual(len(result), 10, "number of results not correct")
self.assertEqual(result[0], "item 0", "first item not correct")
def test_get_list_of_all_values_match(self):
self.add_simple_test_data()
result = get_list_of_all_values(self.test_collection, "$test_txt", unwind=False, match={"test_int": {"$lt": 5}})
self.assertEqual(len(result), 5)
self.assertEqual(result[0], "item 0")
def test_get_list_of_all_values_unwind(self):
self.add_list_test_data()
result = get_list_of_all_values(self.test_collection, "$test_list", unwind=True, match=None)
self.assertEqual(len(result), 4)
self.assertEqual(result[0], "a")
def test_get_objects_and_count_of_occurence(self):
self.add_simple_test_data()
result = get_objects_and_count_of_occurrence(self.test_collection, "$test_txt", unwind=False, match=None)
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment