(self, total_items: int, operation_name: str = "Processing")
| 22 | """Tracks progress and performance metrics for batch operations""" |
| 23 | |
| 24 | def __init__(self, total_items: int, operation_name: str = "Processing"): |
| 25 | self.total_items = total_items |
| 26 | self.operation_name = operation_name |
| 27 | self.processed_items = 0 |
| 28 | self.errors_encountered = 0 |
| 29 | self.start_time = time.time() |
| 30 | self.last_report_time = time.time() |
| 31 | self.report_interval = 10 # Report every 10 seconds |
| 32 | |
| 33 | def update(self, items_processed: int, errors: int = 0): |
| 34 | """Update progress with number of items processed""" |
nothing calls this directly
no outgoing calls
no test coverage detected