Coverage for src/lexigram/admin/services/export/adapters/csv.py: 100%

1from __future__ import annotations 

2 

3import csv 

4from datetime import UTC 

5import io 

6from typing import TYPE_CHECKING, Any 

7 

8from lexigram.admin.services.export.sanitize import sanitize_cell_value 

9from lexigram.admin.services.export.service import IExportBackend 

10 

11if TYPE_CHECKING: 

12 from lexigram.admin.services.export.scheduler import ExportJob 

13 

14 

15class CsvExportBackend(IExportBackend): 

16 """Backend for CSV exports.""" 

17 

18 async def generate_file( 

19 self, 

20 job: ExportJob, 

21 data: list[dict[str, Any]], 

22 storage: Any, 

23 export_dir: str, 

24 ) -> str: 

25 """Export data to CSV format.""" 

26 from datetime import datetime 

27 

28 timestamp = datetime.now(UTC).strftime("%Y%m%d_%H%M%S") 

29 filename = f"{job.resource_name}_export_{timestamp}.csv" 

30 file_path = f"{export_dir}/{filename}" 

31 

32 output = io.StringIO() 

33 

34 if data: 

35 # Use specified columns or all columns 

36 fieldnames = job.columns or list(data[0].keys()) 

37 

38 writer = csv.DictWriter(output, fieldnames=fieldnames) 

39 writer.writeheader() 

40 

41 for row in data: 

42 # Filter to specified columns if provided 

43 if job.columns: 

44 filtered_row = {k: v for k, v in row.items() if k in job.columns} 

45 else: 

46 filtered_row = row 

47 sanitized_row = { 

48 k: sanitize_cell_value(v) for k, v in filtered_row.items() 

49 } 

50 writer.writerow(sanitized_row) 

51 

52 # Write to storage 

53 content = output.getvalue() 

54 await storage.upload( 

55 file_path, content.encode("utf-8"), content_type="text/csv" 

56 ) 

57 

58 return file_path