90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
|
"""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)
|