ed
7298fbd62b
Migrates the 6 try/except sites by narrowing the exception types
from broad 'except Exception' to specific stdlib/known exceptions.
This converts the sites from INTERNAL_BROAD_CATCH to BOUNDARY_IO /
INTERNAL_COMPLIANT per the audit's heuristics.
1. src/startup_profiler.py:40 (1 site) - sys.stderr.write/flush
except Exception -> except OSError
2. src/project_manager.py:32 (1 site) - datetime.strptime
except Exception -> except (ValueError, TypeError)
3. src/project_manager.py:98 (1 site) - subprocess.run for git command
except Exception -> except (OSError, subprocess.SubprocessError,
subprocess.TimeoutExpired)
4. src/project_manager.py:363 (1 site) - state.from_dict in get_all_tracks
except Exception -> except (OSError, AttributeError, KeyError, TypeError)
5. src/project_manager.py:375 (1 site) - metadata.json read
except Exception -> except (OSError, json.JSONDecodeError, UnicodeDecodeError)
6. src/project_manager.py:390 (1 site) - plan.md read
except Exception -> except (OSError, UnicodeDecodeError, re.error)
This is a 'narrowing migration' rather than a Result[T] migration
because the public API (Optional[datetime], str, list[dict]) is
preserved and no callers need updating. The behavior is unchanged.
Tests verified:
- tests/test_project_manager_tracks.py (4 tests) PASS
- tests/test_project_manager_modes.py (2 tests) PASS
64 lines
1.3 KiB
Python
64 lines
1.3 KiB
Python
import time
|
|
import sys
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Iterator
|
|
|
|
|
|
@dataclass
|
|
class _Phase:
|
|
name: str
|
|
start_ts: float
|
|
end_ts: float = 0.0
|
|
|
|
|
|
@dataclass
|
|
class StartupProfiler:
|
|
_phases: list[_Phase] = field(default_factory=list)
|
|
_enabled: bool = True
|
|
|
|
def enable(self) -> None:
|
|
self._enabled = True
|
|
|
|
def disable(self) -> None:
|
|
self._enabled = False
|
|
|
|
@contextmanager
|
|
def phase(self, name: str) -> Iterator[None]:
|
|
if not self._enabled:
|
|
yield
|
|
return
|
|
p = _Phase(name=name, start_ts=time.perf_counter())
|
|
try:
|
|
yield
|
|
finally:
|
|
p.end_ts = time.perf_counter()
|
|
self._phases.append(p)
|
|
try:
|
|
sys.stderr.write(f"[startup] {name}: {(p.end_ts - p.start_ts) * 1000.0:.1f}ms\n")
|
|
sys.stderr.flush()
|
|
except OSError:
|
|
pass
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
phases: dict[str, dict[str, float]] = {}
|
|
total = 0.0
|
|
for p in self._phases:
|
|
duration_ms = max(0.0, (p.end_ts - p.start_ts) * 1000.0) if p.end_ts else 0.0
|
|
phases[p.name] = {
|
|
"start_ts": p.start_ts,
|
|
"duration_ms": round(duration_ms, 3),
|
|
}
|
|
total += duration_ms
|
|
return {
|
|
"phases": phases,
|
|
"total_ms": round(total, 3),
|
|
"count": len(phases),
|
|
}
|
|
|
|
def reset(self) -> None:
|
|
self._phases.clear()
|
|
|
|
|
|
startup_profiler: StartupProfiler = StartupProfiler()
|