Coverage for api/database/text_search/mongo_search.py: 30%

1import re 

2 

3from dbcc import MongoTableEngine 

4from services.opensearch_client import get_ids_query 

5from utils.helper import validate_search_q 

6 

7 

8def build_search_query( 

9 pattern: str, collection_name: str 

10) -> dict[str, dict[str, str | re.Pattern]]: 

11 pattern = validate_search_q(pattern) 

12 if collection_name in ("subsidiaries", "api_key"): 

13 search_field = "name" 

14 else: 

15 search_field = "text_search" 

16 regex = re.compile(pattern, re.RegexFlag.IGNORECASE) 

17 return {search_field: {"$regex": regex}} 

18 

19 

20async def update_query_by_search( 

21 col: MongoTableEngine, search_q: str, query: dict 

22): 

23 search_q_update = await get_ids_query(col, search_q) 

24 if not search_q_update.get("_id", {}).get("$in"): 

25 search_q_update = build_search_query(search_q, col.collection_name) 

26 query.update(search_q_update)