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

1"""Export job models and scheduling types for Lexigram Admin.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from enum import StrEnum 

8from typing import Any 

9 

10 

11class ExportFormat(StrEnum): 

12 """Supported export formats.""" 

13 

14 CSV = "csv" 

15 EXCEL = "xlsx" 

16 JSON = "json" 

17 PDF = "pdf" 

18 

19 

20class ExportStatus(StrEnum): 

21 """Export job status.""" 

22 

23 PENDING = "pending" 

24 PROCESSING = "processing" 

25 COMPLETED = "completed" 

26 FAILED = "failed" 

27 CANCELLED = "cancelled" 

28 

29 

30class ExportSchedule(StrEnum): 

31 """Export scheduling options.""" 

32 

33 IMMEDIATE = "immediate" 

34 HOURLY = "hourly" 

35 DAILY = "daily" 

36 WEEKLY = "weekly" 

37 MONTHLY = "monthly" 

38 

39 

40@dataclass 

41class ExportTemplate: 

42 """Export template configuration.""" 

43 

44 name: str 

45 format: ExportFormat 

46 columns: list[Any] # ExportColumn - defined elsewhere 

47 description: str | None = None 

48 filters: dict[str, Any] = field(default_factory=dict) 

49 sort_by: str | None = None 

50 sort_order: str = "asc" 

51 include_charts: bool = False 

52 chart_configs: list[dict[str, Any]] = field(default_factory=list) 

53 styling: dict[str, Any] = field(default_factory=dict) 

54 

55 

56@dataclass 

57class ExportJob: 

58 """Export job configuration and status.""" 

59 

60 job_id: str 

61 resource_name: str 

62 format: ExportFormat 

63 filters: dict[str, Any] = field(default_factory=dict) 

64 columns: list[str] = field(default_factory=list) 

65 template_name: str | None = None 

66 status: ExportStatus = ExportStatus.PENDING 

67 progress: float = 0.0 

68 total_records: int = 0 

69 processed_records: int = 0 

70 file_path: str | None = None 

71 file_size: int = 0 

72 download_url: str | None = None 

73 error_message: str | None = None 

74 created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) 

75 started_at: datetime | None = None 

76 completed_at: datetime | None = None 

77 user_id: Any | None = None 

78 scheduled_for: datetime | None = None 

79 schedule_type: ExportSchedule = ExportSchedule.IMMEDIATE 

80 email_recipients: list[str] = field(default_factory=list) 

81 metadata: dict[str, Any] = field(default_factory=dict) 

82 

83 

84__all__ = [ 

85 "ExportFormat", 

86 "ExportJob", 

87 "ExportSchedule", 

88 "ExportStatus", 

89 "ExportTemplate", 

90]