Enhanced context object for tracking progress during feast apply operations. This class manages multiple positioned progress bars with fixed-width formatting: 1. Overall progress (position 0) - tracks main phases 2. Phase progress (position 1) - tracks operations within current pha
| 41 | |
| 42 | @dataclass |
| 43 | class ApplyProgressContext: |
| 44 | """ |
| 45 | Enhanced context object for tracking progress during feast apply operations. |
| 46 | |
| 47 | This class manages multiple positioned progress bars with fixed-width formatting: |
| 48 | 1. Overall progress (position 0) - tracks main phases |
| 49 | 2. Phase progress (position 1) - tracks operations within current phase |
| 50 | |
| 51 | Features: |
| 52 | - Fixed-width alignment for perfect visual consistency |
| 53 | - Color-coded progress bars by phase |
| 54 | - Position coordination to prevent overlap |
| 55 | - TTY detection for CI/CD compatibility |
| 56 | """ |
| 57 | |
| 58 | # Core tracking state |
| 59 | current_phase: str = "" |
| 60 | overall_progress: Optional[tqdm] = None |
| 61 | phase_progress: Optional[tqdm] = None |
| 62 | |
| 63 | # Progress tracking |
| 64 | total_phases: int = 3 |
| 65 | completed_phases: int = 0 |
| 66 | tty_available: bool = True |
| 67 | |
| 68 | # Position allocation |
| 69 | OVERALL_POSITION = 0 |
| 70 | PHASE_POSITION = 1 |
| 71 | |
| 72 | def __post_init__(self): |
| 73 | """Initialize TTY detection after dataclass creation.""" |
| 74 | self.tty_available = _PROGRESS_UTILS_AVAILABLE and is_tty_available() |
| 75 | |
| 76 | def start_overall_progress(self): |
| 77 | """Initialize the overall progress bar for apply phases.""" |
| 78 | if not self.tty_available: |
| 79 | return |
| 80 | |
| 81 | if self.overall_progress is None: |
| 82 | try: |
| 83 | self.overall_progress = create_positioned_tqdm( |
| 84 | position=self.OVERALL_POSITION, |
| 85 | description="Applying changes", |
| 86 | total=self.total_phases, |
| 87 | color=get_color_for_phase("overall"), |
| 88 | ) |
| 89 | except (TypeError, AttributeError): |
| 90 | # Handle case where fallback functions don't work as expected |
| 91 | self.overall_progress = None |
| 92 | |
| 93 | def start_phase(self, phase_name: str, operations_count: int = 0): |
| 94 | """ |
| 95 | Start tracking a new phase. |
| 96 | |
| 97 | Args: |
| 98 | phase_name: Human-readable name of the phase |
| 99 | operations_count: Number of operations in this phase (0 for unknown) |
| 100 | """ |
no outgoing calls
no test coverage detected