Coverage for backend/django/core/auxiliary/services/project_folder_queries.py: 100%

1"""Bounded project-folder read models for workspace navigation.""" 

2 

3from django.db import connection 

4 

5from authentication.user.models import User 

6from core.auxiliary.enums.FlowsheetTemplateType import FlowsheetTemplateType 

7from core.auxiliary.models.Flowsheet import Flowsheet 

8from core.auxiliary.models.Project import Project 

9from core.auxiliary.models.ProjectFolder import ProjectFolder 

10 

11 

12def annotate_project_folder_subtree_counts( 

13 *, 

14 folders: list[ProjectFolder], 

15 user: User, 

16) -> list[ProjectFolder]: 

17 """Count regular projects in each displayed folder's active subtree. 

18 

19 The recursive aggregate is rooted only at the bounded response page, so the 

20 API reports the total users expect without materializing either projects or 

21 descendant folders in Python. 

22 """ 

23 

24 if not folders: 

25 return folders 

26 folder_ids = [folder.pk for folder in folders] 

27 quote = connection.ops.quote_name 

28 folder_table = quote(ProjectFolder._meta.db_table) 

29 project_table = quote(Project._meta.db_table) 

30 flowsheet_table = quote(Flowsheet._meta.db_table) 

31 folder_placeholders = ", ".join(["%s"] * len(folder_ids)) 

32 sql = f""" 

33 WITH RECURSIVE folder_tree(root_id, id) AS ( 

34 SELECT id, id 

35 FROM {folder_table} 

36 WHERE id IN ({folder_placeholders}) 

37 AND owner_id = %s 

38 AND NOT is_binned 

39 UNION ALL 

40 SELECT folder_tree.root_id, child.id 

41 FROM {folder_table} child 

42 JOIN folder_tree ON child.parent_id = folder_tree.id 

43 WHERE child.owner_id = %s AND NOT child.is_binned 

44 ) 

45 SELECT folder_tree.root_id, COUNT(flowsheet.id) 

46 FROM folder_tree 

47 LEFT JOIN {project_table} project 

48 ON project.folder_id = folder_tree.id 

49 AND project.owner_id = %s 

50 AND NOT project.is_binned 

51 LEFT JOIN {flowsheet_table} flowsheet 

52 ON flowsheet.id = project.active_flowsheet_id 

53 AND flowsheet.flowsheet_template_type = %s 

54 GROUP BY folder_tree.root_id 

55 """ 

56 with connection.cursor() as cursor: 

57 cursor.execute( 

58 sql, 

59 [ 

60 *folder_ids, 

61 user.pk, 

62 user.pk, 

63 user.pk, 

64 FlowsheetTemplateType.NotTemplate, 

65 ], 

66 ) 

67 counts = dict(cursor.fetchall()) 

68 for folder in folders: 

69 folder.project_count = counts.get(folder.pk, 0) 

70 return folders 

71 

72 

73def list_active_project_folder_ancestors( 

74 *, 

75 folder: ProjectFolder, 

76 user: User, 

77) -> list[ProjectFolder]: 

78 """Load one root-to-current path without materializing sibling subtrees.""" 

79 

80 quote = connection.ops.quote_name 

81 folder_table = quote(ProjectFolder._meta.db_table) 

82 sql = f""" 

83 WITH RECURSIVE ancestors AS ( 

84 SELECT id, parent_id, 0 AS depth 

85 FROM {folder_table} 

86 WHERE id = %s AND owner_id = %s AND NOT is_binned 

87 UNION ALL 

88 SELECT parent.id, parent.parent_id, ancestors.depth + 1 

89 FROM {folder_table} parent 

90 JOIN ancestors ON ancestors.parent_id = parent.id 

91 WHERE parent.owner_id = %s AND NOT parent.is_binned 

92 ) 

93 SELECT id 

94 FROM ancestors 

95 ORDER BY depth DESC 

96 """ 

97 with connection.cursor() as cursor: 

98 cursor.execute(sql, [folder.pk, user.pk, user.pk]) 

99 ancestor_ids = [row[0] for row in cursor.fetchall()] 

100 ancestors_by_id = ProjectFolder.objects.in_bulk(ancestor_ids) 

101 return [ancestors_by_id[folder_id] for folder_id in ancestor_ids]