Add cli/ package with shared debug context
This commit is contained in:
parent
9f38b1ea9c
commit
df0fa41586
2 changed files with 90 additions and 0 deletions
1
cli/__init__.py
Normal file
1
cli/__init__.py
Normal file
|
|
@ -0,0 +1 @@
|
|||
"""Debug CLI package — mirrors web UI interactions for debugging."""
|
||||
89
cli/_context.py
Normal file
89
cli/_context.py
Normal file
|
|
@ -0,0 +1,89 @@
|
|||
"""Shared context for the debug CLI."""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Any
|
||||
|
||||
import jwt
|
||||
|
||||
from database import engine
|
||||
from repositories.user_repository import UserRepository
|
||||
|
||||
|
||||
@dataclass
|
||||
class CliContext:
|
||||
"""Shared state passed through Click context."""
|
||||
user_email: str
|
||||
use_http: bool
|
||||
json_output: bool
|
||||
api_base_url: str
|
||||
|
||||
|
||||
def resolve_user_id(email: str) -> int:
|
||||
"""Get or create a database user by email. Returns the DB user ID."""
|
||||
user_repo = UserRepository(engine)
|
||||
db_user = user_repo.get_user_by_email(email)
|
||||
if db_user is None:
|
||||
db_user = user_repo.create_user(email)
|
||||
if db_user.id is None:
|
||||
raise RuntimeError(f"Failed to resolve user ID for {email}")
|
||||
return db_user.id
|
||||
|
||||
|
||||
def mint_jwt(email: str) -> str:
|
||||
"""Create a self-signed JWT for HTTP mode (passkey-style HS256 token)."""
|
||||
secret = os.getenv("JWT_SECRET", "change-me-in-production")
|
||||
issuer = os.getenv("JWT_ISSUER", "wrongmove")
|
||||
algorithm = os.getenv("JWT_ALGORITHM", "HS256")
|
||||
payload = {
|
||||
"sub": email,
|
||||
"email": email,
|
||||
"name": email,
|
||||
"iss": issuer,
|
||||
"iat": datetime.now(timezone.utc),
|
||||
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
|
||||
}
|
||||
return jwt.encode(payload, secret, algorithm=algorithm)
|
||||
|
||||
|
||||
def get_http_headers(email: str) -> dict[str, str]:
|
||||
"""Build HTTP headers with Authorization bearer token."""
|
||||
token = mint_jwt(email)
|
||||
return {
|
||||
"Authorization": f"Bearer {token}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def output(data: Any, json_mode: bool) -> None:
|
||||
"""Print data in JSON or human-readable format."""
|
||||
if json_mode:
|
||||
print(json.dumps(data, indent=2, default=str))
|
||||
elif isinstance(data, list):
|
||||
if not data:
|
||||
print("No results.")
|
||||
return
|
||||
if isinstance(data[0], dict):
|
||||
keys = list(data[0].keys())
|
||||
print(" ".join(f"{k:<20}" for k in keys))
|
||||
print(" ".join("-" * 20 for _ in keys))
|
||||
for row in data:
|
||||
print(" ".join(f"{str(row.get(k, '')):<20}" for k in keys))
|
||||
else:
|
||||
for item in data:
|
||||
print(f" {item}")
|
||||
elif isinstance(data, dict):
|
||||
for k, v in data.items():
|
||||
print(f" {k}: {v}")
|
||||
else:
|
||||
print(data)
|
||||
|
||||
|
||||
def error_output(message: str, json_mode: bool) -> None:
|
||||
"""Print an error message."""
|
||||
if json_mode:
|
||||
print(json.dumps({"error": message}), file=sys.stderr)
|
||||
else:
|
||||
print(f"Error: {message}", file=sys.stderr)
|
||||
Loading…
Add table
Add a link
Reference in a new issue