Spaces:
Sleeping
Sleeping
File size: 5,335 Bytes
8e0dd55 fffe8f4 8e0dd55 113f6a0 8e0dd55 113f6a0 8e0dd55 113f6a0 8e0dd55 | 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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | import uvicorn
import os
import sys
import logging
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
def setup_adalflow_paths():
"""
Set up adalflow paths for containerized environments like Hugging Face Spaces.
This prevents permission errors when adalflow tries to create directories in restricted locations.
"""
try:
# Check if we're in a containerized environment with restricted filesystem
default_path = os.path.expanduser(os.path.join("~", ".adalflow"))
# Try to create the default path first
try:
os.makedirs(default_path, exist_ok=True)
# Test if we can write to it
test_file = os.path.join(default_path, "test_write")
with open(test_file, "w") as f:
f.write("test")
os.remove(test_file)
# If successful, set the environment variables to this path
os.environ["ADALFLOW_CACHE_DIR"] = default_path
os.environ["ADALFLOW_HOME"] = default_path
# Set XDG variables which many libraries use for cache directories
os.environ["XDG_DATA_HOME"] = default_path
os.environ["XDG_CACHE_HOME"] = default_path
return default_path
except (PermissionError, OSError):
# Fall back to using a writable location in /tmp
fallback_path = "/tmp/adalflow"
try:
os.makedirs(fallback_path, exist_ok=True)
os.environ["ADALFLOW_CACHE_DIR"] = fallback_path
os.environ["ADALFLOW_HOME"] = fallback_path
# Set XDG variables for fallback
os.environ["XDG_DATA_HOME"] = fallback_path
os.environ["XDG_CACHE_HOME"] = fallback_path
# Also override HOME if we're in a restricted environment
if not os.path.exists(os.path.expanduser("~")) or not os.access(os.path.expanduser("~"), os.W_OK):
os.environ["HOME"] = "/tmp"
return fallback_path
except (PermissionError, OSError):
# Last resort: use current working directory
current_dir_path = os.path.join(os.getcwd(), ".adalflow")
os.makedirs(current_dir_path, exist_ok=True)
os.environ["ADALFLOW_CACHE_DIR"] = current_dir_path
os.environ["ADALFLOW_HOME"] = current_dir_path
# Set XDG variables for last resort
os.environ["XDG_DATA_HOME"] = current_dir_path
os.environ["XDG_CACHE_HOME"] = current_dir_path
return current_dir_path
except Exception as e:
print(f"Error setting up adalflow paths: {e}")
# Set a default fallback
fallback_path = "/tmp/adalflow"
os.makedirs(fallback_path, exist_ok=True)
os.environ["ADALFLOW_CACHE_DIR"] = fallback_path
os.environ["ADALFLOW_HOME"] = fallback_path
# Set XDG variables for emergency fallback
os.environ["XDG_DATA_HOME"] = fallback_path
os.environ["XDG_CACHE_HOME"] = fallback_path
# If we can't access home, set it to /tmp
try:
test_home = os.path.expanduser("~")
if not os.path.exists(test_home) or not os.access(test_home, os.W_OK):
os.environ["HOME"] = "/tmp"
except:
os.environ["HOME"] = "/tmp"
return fallback_path
# Set up adalflow paths before any imports that might use adalflow
adalflow_path = setup_adalflow_paths()
print(f"Adalflow path configured to: {adalflow_path}")
print(f"HOME environment variable: {os.environ.get('HOME')}")
print(f"ADALFLOW_HOME: {os.environ.get('ADALFLOW_HOME')}")
print(f"ADALFLOW_CACHE_DIR: {os.environ.get('ADALFLOW_CACHE_DIR')}")
from api.logging_config import setup_logging
# Configure logging
setup_logging()
logger = logging.getLogger(__name__)
# Add the current directory to the path so we can import the api package
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Check for required environment variables
required_env_vars = ['GOOGLE_API_KEY', 'OPENAI_API_KEY']
missing_vars = [var for var in required_env_vars if not os.environ.get(var)]
if missing_vars:
print(f"Warning: Missing environment variables: {', '.join(missing_vars)}")
# Configure Google Generative AI
import google.generativeai as genai
from api.config import GOOGLE_API_KEY
if GOOGLE_API_KEY:
genai.configure(api_key=GOOGLE_API_KEY)
else:
print("Warning: GOOGLE_API_KEY not configured")
if __name__ == "__main__":
# Get port from environment variable or use default
port = int(os.environ.get("PORT", 8001))
# Import the app here to ensure environment variables are set first
from api.api import app
print(f"Starting Streaming API on port {port}")
# Run the FastAPI app with uvicorn
# Disable reload in production/Docker environment
is_development = os.environ.get("NODE_ENV") != "production"
if is_development:
# Prevent infinite logging loop caused by file changes triggering log writes
logging.getLogger("watchfiles.main").setLevel(logging.WARNING)
uvicorn.run(
"api.api:app",
host="0.0.0.0",
port=port,
reload=is_development
)
|