Coverage for rest_api/viewsets_database.py: 27%

1from django.apps import apps 

2from django.db import connection 

3from rest_framework import status 

4from rest_framework import viewsets 

5from rest_framework.decorators import action 

6from rest_framework.response import Response 

7 

8from . import models 

9from .viewsets_sample import SampleFilterMixin 

10from .viewsets_statistics_and_plots import SampleViewSetPlots 

11from .viewsets_statistics_and_plots import SampleViewSetStatistics 

12 

13 

14class DatabaseInfoView( 

15 viewsets.GenericViewSet, 

16 SampleFilterMixin, 

17): 

18 @action(detail=False, methods=["get"]) 

19 def get_database_tables_status(self, request, *args, **kwargs): 

20 """ 

21 Checks if all required database tables are created and connected with Django. 

22 Returns True if all are ready, otherwise False. 

23 """ 

24 # List of all tables from model 

25 # this also include django tables (e.g., auth_user) 

26 expected_tables = {model._meta.db_table for model in apps.get_models()} 

27 

28 # sonar_db = [ 

29 # "sequence", "alignment", 

30 # "alignment2mutation", "annotation_type", 

31 # "replicon", "gene", 

32 # "gene_segment", "lineage", 

33 # "reference", "property", 

34 # "sample", "sample2property", 

35 # "mutation", "mutation2annotation", 

36 # "processing_job", "file_processing", 

37 # "import_log", 

38 # ] 

39 

40 # Retrieve all actual tables from the database 

41 with connection.cursor() as cursor: 

42 cursor.execute( 

43 "SELECT table_name FROM information_schema.tables WHERE table_schema = 'public';" 

44 ) 

45 actual_tables = {row[0] for row in cursor.fetchall()} 

46 

47 # Check for missing tables 

48 missing_tables = set(expected_tables) - actual_tables 

49 

50 if missing_tables: 

51 return Response( 

52 data={"status": False, "missing_tables": list(missing_tables)}, 

53 status=status.HTTP_200_OK, 

54 ) 

55 

56 return Response(data={"status": True}, status=status.HTTP_200_OK) 

57 

58 @action(detail=False, methods=["get"]) # detail=False means it's a list action 

59 def get_database_info(self, request, *args, **kwargs): 

60 result = {} 

61 queryset = self.get_filtered_queryset(request) 

62 statistics = SampleViewSetStatistics().get_statistics() 

63 total_samples = statistics["samples_total"] 

64 metadata_coverage = SampleViewSetPlots().get_metadata_coverage(queryset) 

65 # Add percentages 

66 for key, count in metadata_coverage.items(): 

67 percentage = (count / total_samples) * 100 if total_samples > 0 else 0 

68 metadata_coverage[key] = f"{count} ({percentage:.2f}%)" 

69 

70 result["metadata_coverage"] = metadata_coverage 

71 result.update( 

72 { 

73 "samples_total": total_samples, 

74 "earliest_sampling_date": statistics["first_sample_date"], 

75 "latest_sampling_date": statistics["latest_sample_date"], 

76 } 

77 ) 

78 # Earliest and Latest Genome Import 

79 earliest_genome_import = models.Sample.objects.order_by( 

80 "init_upload_date" 

81 ).first() 

82 latest_genome_import = models.Sample.objects.order_by( 

83 "-init_upload_date" 

84 ).first() 

85 

86 result["earliest_genome_import"] = ( 

87 earliest_genome_import.init_upload_date if earliest_genome_import else None 

88 ) 

89 result["latest_genome_import"] = ( 

90 latest_genome_import.init_upload_date if latest_genome_import else None 

91 ) 

92 # Reference Genomes with organism-specific statistics 

93 result["reference_genomes"] = {} 

94 for reference in models.Reference.objects.all(): 

95 organism = reference.organism 

96 if organism not in result["reference_genomes"]: 

97 result["reference_genomes"][organism] = {} 

98 replicons = models.Replicon.objects.filter( 

99 description__isnull=False, reference__accession=reference.accession 

100 ) 

101 result["reference_genomes"][organism]["replicons"] = [ 

102 f"{replicon.accession} {replicon.description}" for replicon in replicons 

103 ] 

104 result["reference_genomes"][organism]["reference_length"] = [ 

105 reference_replicon.length for reference_replicon in replicons 

106 ] 

107 # Annotated Proteins 

108 annotated_proteins = models.Gene.objects.filter( 

109 cds__gene__symbol__isnull=False, 

110 cds__gene__replicon__reference__accession=reference.accession, 

111 ).values_list("symbol", flat=True) 

112 result["reference_genomes"][organism]["annotated_proteins"] = ", ".join( 

113 sorted(set(annotated_proteins)) 

114 ) 

115 

116 # Unique Sequences for this organism 

117 unique_sequences = ( 

118 models.Sequence.objects.filter( 

119 alignments__replicon__reference__accession=reference.accession 

120 ) 

121 .distinct() 

122 .count() 

123 ) 

124 result["reference_genomes"][organism]["unique_sequences"] = unique_sequences 

125 

126 # Total Genomes (alignments) for this organism 

127 total_genomes = models.Alignment.objects.filter( 

128 replicon__reference__accession=reference.accession 

129 ).count() 

130 result["reference_genomes"][organism]["genomes"] = total_genomes 

131 print(result["reference_genomes"]) 

132 result["database_size"] = self.get_database_size() 

133 result["database_version"] = self.get_database_version() 

134 return Response(data={"detail": result}, status=status.HTTP_200_OK) 

135 

136 def get_database_size(self): 

137 

138 with connection.cursor() as cursor: 

139 cursor.execute( 

140 """ 

141 SELECT pg_size_pretty(pg_database_size(current_database())); 

142 """ 

143 ) 

144 return cursor.fetchone()[0] 

145 

146 def get_database_version(self): 

147 with connection.cursor() as cursor: 

148 cursor.execute("SELECT version();") 

149 return cursor.fetchone()[0]