slaguru666/AIPass
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): DPLAN-0049 wave 4: simplify key resolution, remove dead code

Browse Source 
Co-Authored-By: @api <api@aipass>
This commit is contained in:
AIOSAI
2026-03-22 23:23:32 -07:00
co-authored by @api
parent 19e63ae6df
commit 873df4e062
6 changed files with 47 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/aipass/api/apps/handlers/auth/env.py
+3 -216
View File
@@ -9,14 +9,10 @@
"""
.env File Handler
Manages .env file reading, creation, and parsing.
Searches multiple paths, creates templates, handles env variables.
Manages .env file creation for API credential setup.
Functions:
read_env_file() - Read environment variable from .env files (multi-path search)
read_env_file_dict() - Read all variables from .env file as dictionary
create_env_template() - Create .env template for provider
validate_env_exists() - Check if .env file exists at any search path
"""
# Infrastructure
@@ -24,7 +20,7 @@ from pathlib import Path
import sys
# Standard library
from typing import Optional, Dict, List
from typing import Optional, Dict
# Logging
from aipass.prax import logger
@@ -33,114 +29,6 @@ from aipass.prax import logger
from aipass.api.apps.handlers.json import json_handler
# ==============================================
# CONSTANTS
# ==============================================
# Default .env search paths (in order of priority)
# Navigate: env.py -> auth/ -> handlers/ -> apps/ -> api/
API_ROOT = Path(__file__).resolve().parent.parent.parent.parent
DEFAULT_ENV_PATHS = [
Path.home() / ".secrets" / "aipass" / ".env", # ~/.secrets/aipass/.env (cross-platform standard)
API_ROOT / ".env", # <api_root>/.env
Path.cwd() / ".env", # <cwd>/.env
]
# ==============================================
# ENV FILE READING
# ==============================================
def read_env_file(env_var: str, search_paths: Optional[List[Path]] = None) -> Optional[str]:
"""
Read environment variable from .env files with multi-path search.
Searches multiple .env file locations in order:
1. <api_root>/.env (package-relative)
2. <cwd>/.env (current working directory)
Args:
env_var: Environment variable name to read (e.g., 'OPENROUTER_API_KEY')
search_paths: Optional custom search paths (defaults to DEFAULT_ENV_PATHS)
Returns:
str: Variable value if found, None otherwise
Example:
>>> api_key = read_env_file('OPENROUTER_API_KEY')
>>> if api_key:
... print(f"Found key: {api_key[:20]}...")
"""
paths = search_paths or DEFAULT_ENV_PATHS
for env_file in paths:
if not env_file.exists():
continue
try:
with open(env_file, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith('#'):
continue
# Parse key=value
if '=' in line:
key, value = line.split('=', 1)
if key.strip() == env_var:
# Found env_var in env_file
json_handler.log_operation("env_loaded", {"variable": env_var, "source": str(env_file)})
return value.strip()
except Exception as e:
# Error reading env_file
logger.warning(f"Error reading env file '{env_file}': {e}")
continue
# Variable not found in any .env file
return None
def read_env_file_dict(env_path: Path) -> Dict[str, str]:
"""
Read all environment variables from a .env file as dictionary.
Args:
env_path: Path to specific .env file to read
Returns:
dict: Dictionary of key-value pairs from .env file
Example:
>>> env_vars = read_env_file_dict(Path('api/.env'))
>>> print(env_vars.get('OPENROUTER_API_KEY'))
"""
env_dict = {}
if not env_path.exists():
# .env file not found
return env_dict
try:
with open(env_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
# Skip empty lines and comments
if not line or line.startswith('#'):
continue
# Parse key=value
if '=' in line:
key, value = line.split('=', 1)
env_dict[key.strip()] = value.strip()
# Read variables from env_path
return env_dict
except Exception as e:
# Error reading env_path
logger.error(f"Error reading env file '{env_path}': {e}")
return env_dict
# ==============================================
# ENV FILE CREATION
# ==============================================
@@ -206,6 +94,7 @@ OPENAI_API_KEY=sk-your-openai-key-here
# Created .env template
logger.info(f"Created .env template at {env_path}")
json_handler.log_operation("env_template_created", {"path": str(env_path), "provider": provider})
return True
except Exception as e:
@@ -214,105 +103,3 @@ OPENAI_API_KEY=sk-your-openai-key-here
return False
def create_custom_env_template(variables: Dict[str, str], target_path: Path,
header: Optional[str] = None) -> bool:
"""
Create custom .env template with specific variables.
Args:
variables: Dictionary of variable names to placeholder values
target_path: Path where .env file should be created
header: Optional custom header comment
Returns:
bool: True if successful
Example:
>>> vars = {
... 'DATABASE_URL': 'postgresql://localhost/mydb',
... 'SECRET_KEY': 'your-secret-key-here'
... }
>>> create_custom_env_template(vars, Path('/path/to/.env'))
"""
# Don't overwrite existing file
if target_path.exists():
# .env file already exists
return True
try:
# Ensure parent directory exists
target_path.parent.mkdir(parents=True, exist_ok=True)
# Build template content
content_lines = []
# Add header
if header:
content_lines.append(f"# {header}")
else:
content_lines.append("# Environment Variables")
content_lines.append("")
# Add variables
for key, value in variables.items():
content_lines.append(f"{key}={value}")
content = "\n".join(content_lines) + "\n"
# Write file
with open(target_path, 'w', encoding='utf-8') as f:
f.write(content)
# Created custom .env template
logger.info(f"Created custom .env template at {target_path}")
return True
except Exception as e:
# Failed to create custom .env template
logger.error(f"Failed to create custom .env template: {e}")
return False
# ==============================================
# VALIDATION
# ==============================================
def validate_env_exists(search_paths: Optional[List[Path]] = None) -> Optional[Path]:
"""
Check if .env file exists at any search path.
Args:
search_paths: Optional custom search paths (defaults to DEFAULT_ENV_PATHS)
Returns:
Path: First found .env file path, or None if none exist
Example:
>>> env_path = validate_env_exists()
>>> if env_path:
... print(f"Found .env at {env_path}")
"""
paths = search_paths or DEFAULT_ENV_PATHS
for env_path in paths:
if env_path.exists():
# Found .env file
return env_path
# No .env file found in search paths
return None
def get_env_search_paths() -> List[Path]:
"""
Get list of default .env search paths.
Returns:
list: List of Path objects for .env search locations
Example:
>>> paths = get_env_search_paths()
>>> for p in paths:
... print(p)
"""
return DEFAULT_ENV_PATHS.copy()
src/aipass/api/apps/handlers/auth/keys.py
+43 -58
View File
@@ -10,13 +10,12 @@
API Key Management Handler
Handles API key retrieval and validation for multiple providers.
Uses fallback chain: config → env → .env files.
Keys are read from config JSON or directly from ~/.secrets/aipass/.env.
Functions:
get_api_key() - Get validated API key with fallback chain
get_api_key() - Get validated API key
validate_key() - Validate key format for provider
get_key_from_config() - Retrieve key from config JSON
get_key_from_env() - Retrieve key from environment variable
get_validation_rules() - Get provider-specific validation rules
"""
@@ -25,15 +24,11 @@ from pathlib import Path
import sys
# Standard library
import os
from typing import Optional, Dict, Any
# Logging
from aipass.prax import logger
# Internal handlers
from aipass.api.apps.handlers.auth.env import read_env_file
# JSON handler
from aipass.api.apps.handlers.json import json_handler
@@ -73,12 +68,11 @@ VALIDATION_RULES = {
def get_api_key(provider: str = "openrouter") -> Optional[str]:
"""
Get validated API key for provider with fallback chain.
Get validated API key for provider.
Fallback order:
Sources (in order):
1. Config JSON file (api_json/api_connect_config.json)
2. Environment variable
3. .env file (multi-path search)
2. Secrets file (~/.secrets/aipass/.env)
Args:
provider: Provider name (default: 'openrouter')
@@ -97,23 +91,13 @@ def get_api_key(provider: str = "openrouter") -> Optional[str]:
# 1. Try config file
key = get_key_from_config(provider)
if key and validate_key(key, provider):
# Using key from config
source = "config"
# 2. Try environment variable
# 2. Try secrets file
if not source:
key = get_key_from_env(provider)
key = _read_key_from_secrets(provider)
if key and validate_key(key, provider):
# Using key from environment
source = "env"
# 3. Try .env file
if not source:
env_var = f"{provider.upper()}_API_KEY"
key = read_env_file(env_var)
if key and validate_key(key, provider):
# Using key from .env file
source = "dotenv"
source = "secrets"
if source:
json_handler.log_operation("key_retrieved", {"provider": provider, "source": source})
@@ -123,11 +107,41 @@ def get_api_key(provider: str = "openrouter") -> Optional[str]:
return None
except Exception as e:
# Failed to get key
logger.error(f"Failed to get API key for provider '{provider}': {e}")
return None
def _read_key_from_secrets(provider: str) -> Optional[str]:
"""
Read API key directly from ~/.secrets/aipass/.env.
Args:
provider: Provider name (e.g., 'openrouter')
Returns:
str: API key value or None if not found
"""
secrets_path = Path.home() / ".secrets" / "aipass" / ".env"
if not secrets_path.exists():
return None
try:
env_var = f"{provider.upper()}_API_KEY"
with open(secrets_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
if '=' in line:
key, value = line.split('=', 1)
if key.strip() == env_var:
return value.strip()
return None
except Exception as e:
logger.warning(f"Error reading secrets file: {e}")
return None
def get_key_from_config(provider: str) -> Optional[str]:
"""
Retrieve API key from config JSON file.
@@ -171,31 +185,6 @@ def get_key_from_config(provider: str) -> Optional[str]:
return None
def get_key_from_env(provider: str) -> Optional[str]:
"""
Retrieve API key from environment variable.
Checks os.environ for {PROVIDER}_API_KEY.
Args:
provider: Provider name (e.g., 'openrouter')
Returns:
str: API key from environment or None if not found
Example:
>>> key = get_key_from_env('openrouter')
>>> # Checks OPENROUTER_API_KEY env variable
"""
env_var = f"{provider.upper()}_API_KEY"
key = os.getenv(env_var)
if key:
# Found key in environment variable
return key
return None
# ==============================================
# KEY VALIDATION
@@ -295,16 +284,12 @@ def diagnose_key(provider: str = "openrouter") -> str:
source = "config"
if not key:
key = get_key_from_env(provider)
source = "env"
key = _read_key_from_secrets(provider)
source = "secrets"
if not key:
env_var = f"{provider.upper()}_API_KEY"
key = read_env_file(env_var)
source = "dotenv"
if not key:
return "No API key found in any source (config, environment, .env file)"
secrets_path = Path.home() / ".secrets" / "aipass" / ".env"
return f"API key for {provider} not found. Expected at {secrets_path}. Run drone @api setup to configure."
# Key exists but failed validation — explain why
key = key.strip()
src/aipass/api/apps/handlers/openrouter/provision.py
-1
View File
@@ -167,7 +167,6 @@ def provision_json_folder(json_folder: Path) -> bool:
json_folder.mkdir(parents=True, exist_ok=True)
logger.info(f"Created JSON folder: {json_folder}")
logger.info(f"Created JSON folder: {json_folder}")
return True
except Exception as e:
src/aipass/api/apps/handlers/usage/aggregation.py
-48
View File
@@ -242,51 +242,3 @@ def calculate_totals(usage_data: List[Dict]) -> Dict[str, float]:
}
def get_model_breakdown(caller: Optional[str] = None) -> Dict[str, Dict[str, int]]:
"""
Calculate model usage breakdown by caller or globally
Args:
caller: Optional caller name to filter by (None = all callers)
Returns:
Dict of {model_name: {"requests": count}}
Returns empty dict {} if no data found
"""
try:
data_path = API_JSON_DIR / DATA_FILE
if not data_path.exists():
logger.info(f"[{MODULE_NAME}] No model breakdown data file found")
return {}
with open(data_path, 'r', encoding='utf-8') as f:
data = json.load(f)
if not data or "data" not in data:
logger.info(f"[{MODULE_NAME}] No model breakdown data available")
return {}
model_stats = {}
usage_by_caller = data["data"].get("usage_by_caller", {})
if caller:
# Single caller breakdown
caller_data = usage_by_caller.get(caller, {})
models_used = caller_data.get("models_used", {})
for model, count in models_used.items():
model_stats[model] = {"requests": count}
else:
# Global breakdown across all callers
for caller_name, caller_data in usage_by_caller.items():
models_used = caller_data.get("models_used", {})
for model, count in models_used.items():
if model not in model_stats:
model_stats[model] = {"requests": 0}
model_stats[model]["requests"] += count
logger.info(f"[{MODULE_NAME}] Retrieved model breakdown: {len(model_stats)} models")
return model_stats
except Exception as e:
logger.error(f"[{MODULE_NAME}] Failed to get model breakdown: {e}")
return {}
src/aipass/api/apps/handlers/usage/cleanup.py
-64
View File
@@ -125,70 +125,6 @@ def _identify_old_generations(generation_tracking: Dict, cutoff_date: datetime)
return old_generations
def cleanup_daily_totals(data_file_path: Path, retention_days: int = 90) -> int:
"""Remove daily total entries older than retention period."""
try:
cutoff_date = (datetime.now() - timedelta(days=retention_days)).date()
data = _read_json(data_file_path)
if not data:
return 0
data_content = data.get("data", data)
old_dates = []
daily_totals = data_content.get("daily_totals", {})
for date_str in daily_totals.keys():
try:
date_obj = datetime.fromisoformat(date_str).date()
if date_obj < cutoff_date:
old_dates.append(date_str)
except (ValueError, TypeError) as e:
logger.warning(f"Invalid date format '{date_str}', marking for cleanup: {e}")
old_dates.append(date_str)
if not old_dates:
return 0
for date_str in old_dates:
del data_content["daily_totals"][date_str]
if "data" in data:
data["data"] = data_content
data["timestamp"] = datetime.now().isoformat()
_write_json(data_file_path, data)
logger.info(f"Cleaned up {len(old_dates)} daily total entries")
logger.info(f"Cleaned up {len(old_dates)} daily total entries older than {retention_days} days")
return len(old_dates)
except Exception as e:
# logger.error(f"Daily totals cleanup failed: {e}")
logger.error(f"Daily totals cleanup failed: {e}")
raise
def auto_cleanup(data_file_path: Path, config: Optional[Dict] = None) -> Dict[str, int]:
"""Perform automatic cleanup based on configuration."""
try:
gen_retention = config.get("cleanup_old_data_days", 30) if config else 30
daily_retention = config.get("cleanup_daily_totals_days", 90) if config else 90
generations_removed = cleanup_old_data(data_file_path, gen_retention)
daily_totals_removed = cleanup_daily_totals(data_file_path, daily_retention)
logger.info(f"Auto cleanup: {generations_removed} generations, {daily_totals_removed} daily totals removed")
return {
"generations_removed": generations_removed,
"daily_totals_removed": daily_totals_removed
}
except Exception as e:
# logger.error(f"Auto cleanup failed: {e}")
logger.error(f"Auto cleanup failed: {e}")
raise
def get_cleanup_stats(data_file_path: Path) -> Dict[str, int]:
"""Get statistics about data that could be cleaned up."""
src/aipass/api/apps/modules/api_key.py
+1 -1
View File
@@ -37,7 +37,7 @@ def print_introspection():
console.print("[cyan]Connected Handlers:[/cyan]")
console.print(" • api.apps.handlers.auth.keys")
console.print(" • api.apps.handlers.auth.env")
console.print(" • api.apps.handlers.auth.env (template creation)")
console.print(" • api.apps.handlers.config.provider")
console.print(" • api.apps.handlers.json.json_handler")
console.print()