Coverage for app\services\export_service.py: 25%

1import csv 

2import os 

3import json 

4import tempfile 

5from datetime import datetime, UTC 

6from typing import List, Dict, Any, Optional 

7from sqlalchemy.orm import Session 

8try: 

9 import pandas as pd 

10except ImportError: 

11 pd = None 

12from reportlab.lib import colors 

13from reportlab.lib.pagesizes import letter, A4, landscape 

14from reportlab.platypus import SimpleDocTemplate, Table, TableStyle, Paragraph, Spacer 

15from reportlab.lib.styles import getSampleStyleSheet 

16 

17from ..db import User, School, AuditLog, SchoolMetric, DataExport 

18 

19class ExportService: 

20 def __init__(self, db: Session): 

21 self.db = db 

22 self.temp_dir = tempfile.gettempdir() 

23 if not os.path.exists(self.temp_dir): 

24 os.makedirs(self.temp_dir, exist_ok=True) 

25 

26 def generate_export(self, export_record: DataExport) -> str: 

27 """ 

28 Generates the export file and returns the file path. 

29 """ 

30 data = self._fetch_data(export_record.export_type, export_record.filters) 

31 

32 filename = f"export_{export_record.export_type}_{datetime.now(UTC).strftime('%Y%m%d%H%M%S')}" 

33 file_path = os.path.join(self.temp_dir, f"{filename}.{export_record.format}") 

34 

35 if export_record.format == "csv": 

36 self._write_csv(data, file_path) 

37 elif export_record.format == "xlsx": 

38 self._write_excel(data, file_path) 

39 elif export_record.format == "pdf": 

40 self._write_pdf(data, file_path, title=f"Export: {export_record.export_type}") 

41 else: 

42 raise ValueError(f"Unsupported format: {export_record.format}") 

43 

44 return file_path 

45 

46 def _fetch_data(self, export_type: str, filters: Optional[Dict[str, Any]]) -> List[Dict[str, Any]]: 

47 if export_type == "users": 

48 query = self.db.query(User) 

49 # Apply filters if needed 

50 users = query.all() 

51 return [ 

52 { 

53 "id": u.id, 

54 "email": u.email, 

55 "full_name": u.full_name, 

56 "role": u.role.value if u.role else None, 

57 "status": u.status.value if u.status else None, 

58 "created_at": u.created_at.isoformat() if u.created_at else None 

59 } 

60 for u in users 

61 ] 

62 elif export_type == "schools": 

63 query = self.db.query(School) 

64 schools = query.all() 

65 return [ 

66 { 

67 "id": s.id, 

68 "name": s.name, 

69 "status": s.status, 

70 "director": s.director_name, 

71 "created_at": s.created_at.isoformat() if s.created_at else None 

72 } 

73 for s in schools 

74 ] 

75 elif export_type == "audit_logs": 

76 query = self.db.query(AuditLog).order_by(AuditLog.created_at.desc()).limit(1000) 

77 logs = query.all() 

78 return [ 

79 { 

80 "id": l.id, 

81 "actor_id": l.actor_id, 

82 "action": l.action.value if l.action else None, 

83 "entity": l.entity, 

84 "entity_id": l.entity_id, 

85 "created_at": l.created_at.isoformat() if l.created_at else None 

86 } 

87 for l in logs 

88 ] 

89 else: 

90 return [] 

91 

92 def _write_csv(self, data: List[Dict[str, Any]], file_path: str): 

93 if not data: 

94 with open(file_path, 'w', newline='') as f: 

95 pass 

96 return 

97 

98 keys = data[0].keys() 

99 with open(file_path, 'w', newline='', encoding='utf-8') as f: 

100 dict_writer = csv.DictWriter(f, fieldnames=keys) 

101 dict_writer.writeheader() 

102 dict_writer.writerows(data) 

103 

104 def _write_excel(self, data: List[Dict[str, Any]], file_path: str): 

105 if pd is None: 

106 raise ImportError("pandas is not installed. Excel export is unavailable.") 

107 

108 if not data: 

109 df = pd.DataFrame() 

110 else: 

111 df = pd.DataFrame(data) 

112 

113 df.to_excel(file_path, index=False) 

114 

115 def _write_pdf(self, data: List[Dict[str, Any]], file_path: str, title: str = "Report"): 

116 doc = SimpleDocTemplate(file_path, pagesize=landscape(A4)) 

117 elements = [] 

118 styles = getSampleStyleSheet() 

119 

120 # Title 

121 elements.append(Paragraph(title, styles['Title'])) 

122 elements.append(Spacer(1, 12)) 

123 

124 if not data: 

125 elements.append(Paragraph("No data available", styles['Normal'])) 

126 doc.build(elements) 

127 return 

128 

129 # Table Data 

130 keys = list(data[0].keys()) 

131 table_data = [keys] # Header 

132 

133 for item in data: 

134 row = [str(item.get(key, "")) for key in keys] 

135 table_data.append(row) 

136 

137 # Table Style 

138 table = Table(table_data) 

139 style = TableStyle([ 

140 ('BACKGROUND', (0, 0), (-1, 0), colors.grey), 

141 ('TEXTCOLOR', (0, 0), (-1, 0), colors.whitesmoke), 

142 ('ALIGN', (0, 0), (-1, -1), 'CENTER'), 

143 ('FONTNAME', (0, 0), (-1, 0), 'Helvetica-Bold'), 

144 ('BOTTOMPADDING', (0, 0), (-1, 0), 12), 

145 ('BACKGROUND', (0, 1), (-1, -1), colors.beige), 

146 ('GRID', (0, 0), (-1, -1), 1, colors.black), 

147 ]) 

148 table.setStyle(style) 

149 

150 elements.append(table) 

151 doc.build(elements)