n0mad1k/coe-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix type hints and resource management in coe_mirror

Browse Source 
- 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:
n0mad1k
2026-06-14 14:40:14 -04:00
parent c322488869
commit d9e01961ff
1 changed files with 35 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
coe_mirror.py
+35 -36
View File
@@ -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,44 +415,43 @@ 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()
session.headers.update({"User-Agent": USER_AGENT})
with requests.Session() as session:
session.headers.update({"User-Agent": USER_AGENT})
encoded_room = quote(MATRIX_ROOM, safe="")
url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/rooms/{encoded_room}/send/m.room.message"
body = {"msgtype": "m.text", "body": message}
encoded_room = quote(MATRIX_ROOM, safe="")
url = f"{MATRIX_HOMESERVER}/_matrix/client/r0/rooms/{encoded_room}/send/m.room.message"
body = {"msgtype": "m.text", "body": message}
for attempt in (1, 2):
token = _get_matrix_token(session, force_refresh=(attempt == 2))
if not token:
logger.warning("matrix send aborted: no token (attempt %d)", attempt)
return False
try:
resp = session.post(
url,
json=body,
headers={"Authorization": f"Bearer {token}"},
timeout=MATRIX_TIMEOUT,
)
if resp.status_code == 200:
logger.info("matrix sent: %s", message[:80])
return True
if resp.status_code == 401 and attempt == 1:
logger.info("matrix 401 on send; refreshing token and retrying")
continue
logger.warning("matrix send failed: HTTP %d body=%s",
resp.status_code, resp.text[:200])
return False
except requests.RequestException as exc:
logger.warning("matrix send network error: %s", exc)
return False
except Exception as exc:
logger.warning("matrix send unexpected error: %s", exc)
return False
return False
for attempt in (1, 2):
token = _get_matrix_token(session, force_refresh=(attempt == 2))
if not token:
logger.warning("matrix send aborted: no token (attempt %d)", attempt)
return False
try:
resp = session.post(
url,
json=body,
headers={"Authorization": f"Bearer {token}"},
timeout=MATRIX_TIMEOUT,
)
if resp.status_code == 200:
logger.info("matrix sent: %s", message[:80])
return True
if resp.status_code == 401 and attempt == 1:
logger.info("matrix 401 on send; refreshing token and retrying")
continue
logger.warning("matrix send failed: HTTP %d body=%s",
resp.status_code, resp.text[:200])
return False
except requests.RequestException as exc:
logger.warning("matrix send network error: %s", exc)
return False
except Exception as exc:
logger.warning("matrix send unexpected error: %s", exc)
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