()
| 73 | |
| 74 | def report_input_forever(): |
| 75 | def task_manager(): |
| 76 | minutes = 180 |
| 77 | check_interval_seconds = 60 * minutes |
| 78 | |
| 79 | while True: |
| 80 | print("[input report monitor] heartbeat") |
| 81 | with sessionMaker.session_scope() as session: |
| 82 | update_input = Update_Input( |
| 83 | session = session |
| 84 | ) |
| 85 | |
| 86 | update_input.report_input_list_recent( |
| 87 | include_within_x_minutes = minutes) |
| 88 | |
| 89 | time.sleep(check_interval_seconds) |
| 90 | |
| 91 | t = threading.Timer(0, task_manager) |
| 92 | t.daemon = True |
nothing calls this directly
no test coverage detected