40 lines
1.1 KiB
Python
Executable File
40 lines
1.1 KiB
Python
Executable File
"""Definition of function to get visual communication from database."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from pymongo.collection import Collection
|
|
|
|
from shared.database import NoDocumentFoundException
|
|
from shared.database import VisualCommunication
|
|
|
|
|
|
def get_visual_communication(
|
|
collection: Collection,
|
|
with_annotation: bool = False,
|
|
) -> VisualCommunication:
|
|
"""Get a random visual communication from the database."""
|
|
query = {}
|
|
if with_annotation:
|
|
query['annotation'] = {'$ne': None}
|
|
else:
|
|
query['annotation'] = {'$eq': None}
|
|
data = collection.aggregate(
|
|
pipeline=[
|
|
{
|
|
'$match': query, # find using filters
|
|
},
|
|
{
|
|
'$sample': {
|
|
'size': 1, # get one random
|
|
},
|
|
},
|
|
],
|
|
)
|
|
data_list = list(data) # read data from cursor object
|
|
if len(data_list) == 0:
|
|
raise NoDocumentFoundException()
|
|
vis_com = VisualCommunication.model_validate(data_list[0])
|
|
logging.debug('finished')
|
|
return vis_com
|