Spaces:
Running
Running
File size: 3,613 Bytes
d0dd276 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
from fastapi import HTTPException, Header, Depends
from fastapi.security import APIKeyHeader
from typing import Optional
from app.config import settings
import app.vertex.config as config
import os
import json
from app.utils.logging import vertex_log
# API Key security scheme
api_key_header = APIKeyHeader(name="Authorization", auto_error=False)
# Function to validate API key
def validate_api_key(api_key_to_validate: str) -> bool:
return True
# Dependency for API key validation
async def get_api_key(authorization: Optional[str] = Header(None)):
if authorization is None:
raise HTTPException(
status_code=401,
detail="Missing API key. Please include 'Authorization: Bearer YOUR_API_KEY' header."
)
# Check if the header starts with "Bearer "
if not authorization.startswith("Bearer "):
raise HTTPException(
status_code=401,
detail="Invalid API key format. Use 'Authorization: Bearer YOUR_API_KEY'"
)
# Extract the API key
api_key = authorization.replace("Bearer ", "")
# Validate the API key
if not validate_api_key(api_key): # Call local validate_api_key
raise HTTPException(
status_code=401,
detail="Invalid API key"
)
return api_key
def validate_settings():
"""Validate settings for Vertex API access."""
# 检查API key
api_key = None
if hasattr(settings, 'API_KEY') and settings.API_KEY:
api_key = settings.API_KEY
else:
api_key = config.API_KEY
if not api_key:
vertex_log('warning', "API key is not set. Some functionality may be limited.")
# 检查Google credentials JSON
google_credentials_json = None
if hasattr(settings, 'GOOGLE_CREDENTIALS_JSON') and settings.GOOGLE_CREDENTIALS_JSON:
google_credentials_json = settings.GOOGLE_CREDENTIALS_JSON
else:
google_credentials_json = config.GOOGLE_CREDENTIALS_JSON
if google_credentials_json:
try:
# 尝试解析JSON确保其有效
json.loads(google_credentials_json)
vertex_log('info', "Google Credentials JSON is valid")
except json.JSONDecodeError:
vertex_log('error', "Google Credentials JSON is not valid JSON. Please check the format.")
return False
# 检查project ID
project_id = None
if hasattr(settings, 'PROJECT_ID') and settings.PROJECT_ID:
project_id = settings.PROJECT_ID
else:
project_id = config.PROJECT_ID
if not project_id:
vertex_log('warning', "Vertex AI Project ID is not set. Required for non-API key methods.")
# 检查location
location = None
if hasattr(settings, 'LOCATION') and settings.LOCATION:
location = settings.LOCATION
else:
location = config.LOCATION
if not location:
vertex_log('warning', "Vertex AI Location is not set, using default: us-central1")
# 验证凭证目录
credentials_dir = None
if hasattr(settings, 'CREDENTIALS_DIR') and settings.CREDENTIALS_DIR:
credentials_dir = settings.CREDENTIALS_DIR
else:
credentials_dir = config.CREDENTIALS_DIR
if not os.path.exists(credentials_dir):
try:
os.makedirs(credentials_dir, exist_ok=True)
vertex_log('info', f"Created credentials directory at: {credentials_dir}")
except Exception as e:
vertex_log('error', f"Failed to create credentials directory: {e}")
return False
return True |