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 """Search manager.""" 

10 

11 def run_search(self, search, user): 

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

13 

14 if search.layers: 

15 for layer in search.layers: 

16 for finder in layer.finders: 

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

18 if search.limit and len(results) > search.limit: 

19 return results[: search.limit] 

20 

21 if search.project: 

22 for finder in search.project.finders: 

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

24 if search.limit and len(results) > search.limit: 

25 return results[: search.limit] 

26 

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

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

29 if search.limit and len(results) > search.limit: 

30 return results[: search.limit] 

31 

32 return results 

33 

34 def _run_search( 

35 self, 

36 search: gws.SearchQuery, 

37 user: gws.User, 

38 finder: gws.Finder, 

39 layer: Optional[gws.Layer], 

40 results, 

41 ): 

42 """Run a single finder and append its features to the results.""" 

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

44 

45 if not user.can_use(finder): 

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

47 return 

48 

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

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

51 return 

52 

53 try: 

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

55 except Exception: 

56 gws.log.exception('SEARCH_FAILED') 

57 return 

58 

59 for feature in features: 

60 if finder.category: 

61 feature.category = finder.category 

62 elif finder.title: 

63 feature.category = finder.title 

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

65 feature.category = layer.title 

66 

67 if search.bounds: 

68 feature.transform_to(search.bounds.crs) 

69 

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

71 if search.limit and len(results) > search.limit: 

72 break 

73 

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