Fix type hints and resource management in coe_mirror
- Add precise type hints: Optional[Tuple[str, str]] for _fetch_creds(), List[Tuple[str, str, str]] for notify() - Wrap requests.Session in context manager for proper cleanup - Remove unreachable return False after retry loop in _send_matrix()
This commit is contained in:
+4
-5
@@ -17,7 +17,7 @@ import zipfile
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from fcntl import flock, LOCK_EX, LOCK_NB
|
||||
from typing import Optional
|
||||
from typing import List, Optional, Tuple
|
||||
from urllib.parse import quote
|
||||
|
||||
import requests
|
||||
@@ -340,7 +340,7 @@ def _write_token_cache(access_token: str, expires_at_iso: str) -> None:
|
||||
logger.warning("token cache write failed: %s", exc)
|
||||
|
||||
|
||||
def _fetch_creds() -> Optional[tuple]:
|
||||
def _fetch_creds() -> Optional[Tuple[str, str]]:
|
||||
"""Fetch (user, password) from Infisical via creds. Return None on failure."""
|
||||
try:
|
||||
user = subprocess.run(
|
||||
@@ -415,7 +415,7 @@ def _send_matrix(message: str) -> bool:
|
||||
Non-fatal: any failure logs WARN and returns False. Never raises.
|
||||
Implements one 401 retry with token refresh.
|
||||
"""
|
||||
session = requests.Session()
|
||||
with requests.Session() as session:
|
||||
session.headers.update({"User-Agent": USER_AGENT})
|
||||
|
||||
encoded_room = quote(MATRIX_ROOM, safe="")
|
||||
@@ -449,10 +449,9 @@ def _send_matrix(message: str) -> bool:
|
||||
except Exception as exc:
|
||||
logger.warning("matrix send unexpected error: %s", exc)
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
def notify(summary: list) -> None:
|
||||
def notify(summary: List[Tuple[str, str, str]]) -> None:
|
||||
"""Build a Matrix message and POST it. Non-fatal."""
|
||||
if not summary:
|
||||
return
|
||||
|
||||
Reference in New Issue
Block a user