Coverage for gws-app/gws/base/search/manager.py: 62%

1"""Search manager.""" 

2 

3from typing import Optional 

4 

5import gws 

6 

7 

8class Object(gws.SearchManager): 

9 def run_search(self, search, user): 

10 results: list[gws.SearchResult] = [] 

11 

12 if search.layers: 

13 for layer in search.layers: 

14 for finder in layer.finders: 

15 self._run_search(search, user, finder, layer, results) 

16 if len(results) > search.limit: 

17 return results[: search.limit] 

18 

19 if search.project: 

20 for finder in search.project.finders: 

21 self._run_search(search, user, finder, None, results) 

22 if len(results) > search.limit: 

23 return results[: search.limit] 

24 

25 for finder in self.root.app.finders: 

26 self._run_search(search, user, finder, None, results) 

27 if len(results) > search.limit: 

28 return results[: search.limit] 

29 

30 return results 

31 

32 def _run_search( 

33 self, 

34 search: gws.SearchQuery, 

35 user: gws.User, 

36 finder: gws.Finder, 

37 layer: Optional[gws.Layer], 

38 results, 

39 ): 

40 gws.log.debug(f'SEARCH_BEGIN: {finder=} {layer=}') 

41 

42 if not user.can_use(finder): 

43 gws.log.debug('SEARCH_END: no access') 

44 return 

45 

46 if not finder.can_run(search, user): 

47 gws.log.debug(f'SEARCH_END: cannot run') 

48 return 

49 

50 try: 

51 features: list[gws.Feature] = finder.run(search, user, layer) or [] 

52 except Exception: 

53 gws.log.exception('SEARCH_FAILED') 

54 return 

55 

56 for feature in features: 

57 if finder.category: 

58 feature.category = finder.category 

59 elif finder.title: 

60 feature.category = finder.title 

61 elif not feature.category and layer and layer.title: 

62 feature.category = layer.title 

63 

64 if search.bounds: 

65 feature.transform_to(search.bounds.crs) 

66 

67 results.append(gws.SearchResult(feature=feature, layer=layer, finder=finder)) 

68 if len(results) > search.limit: 

69 break 

70 

71 gws.log.debug(f'SEARCH_END, found={len(features)}')