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

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.service import IExportBackend 

9 

10if TYPE_CHECKING: 

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

12 

13 

14class CsvExportBackend(IExportBackend): 

15 """Backend for CSV exports.""" 

16 

17 async def generate_file( 

18 self, 

19 job: ExportJob, 

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

21 storage: Any, 

22 export_dir: str, 

23 ) -> str: 

24 """Export data to CSV format.""" 

25 from datetime import datetime 

26 

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

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

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

30 

31 output = io.StringIO() 

32 

33 if data: 

34 # Use specified columns or all columns 

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

36 

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

38 writer.writeheader() 

39 

40 for row in data: 

41 # Filter to specified columns if provided 

42 if job.columns: 

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

44 else: 

45 filtered_row = row 

46 writer.writerow(filtered_row) 

47 

48 # Write to storage 

49 content = output.getvalue() 

50 await storage.upload( 

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

52 ) 

53 

54 return file_path