Spaces:
Running
on
Zero
Running
on
Zero
File size: 4,795 Bytes
0ccf2f0 |
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 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 |
#!/usr/bin/env python3
"""Direct server starter for Warbler CDA API Server.
This script provides a simple way to start the FastAPI server with uvicorn.
It includes basic debugging output and error handling.
"""
import argparse
import logging
import os
import sys
import traceback
from dataclasses import dataclass
from typing import Optional
from urllib.parse import urlparse
import uvicorn
from warbler_cda.api.service import app
# Constants
DEFAULT_HOST = "127.0.0.1"
DEFAULT_PORT = 8000
DEFAULT_LOG_LEVEL = "info"
SEPARATOR_LENGTH = 40
@dataclass
class ServerConfig:
"""Configuration for the server."""
host: str
port: int
log_level: str
reload: bool
def __post_init__(self) -> None:
"""Validate configuration values."""
if not (1 <= self.port <= 65535):
raise ValueError(f"Port must be between 1 and 65535, got {self.port}")
# Basic host validation - accept localhost, IP addresses, or domain names
if not self.host or len(self.host) > 253:
raise ValueError(f"Invalid host: {self.host}")
# Check if it's a valid hostname/IP
try:
urlparse(f"http://{self.host}")
except ValueError:
raise ValueError(f"Invalid host format: {self.host}")
# Validate log level
valid_levels = ["critical", "error", "warning", "info", "debug", "trace"]
if self.log_level.lower() not in valid_levels:
raise ValueError(f"Log level must be one of {valid_levels}, got {self.log_level}")
def parse_args() -> ServerConfig:
"""Parse command line arguments and return validated configuration."""
parser = argparse.ArgumentParser(
description="Start the Warbler CDA API server",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
"--host", default=os.getenv("HOST", DEFAULT_HOST),
help="Host to bind the server to"
)
parser.add_argument(
"--port", "-p", type=int, default=int(os.getenv("PORT", str(DEFAULT_PORT))),
help="Port to bind the server to"
)
parser.add_argument(
"--log-level", "-l",
choices=["critical", "error", "warning", "info", "debug", "trace"],
default=os.getenv("LOG_LEVEL", DEFAULT_LOG_LEVEL).lower(),
help="Uvicorn log level"
)
parser.add_argument(
"--reload", action="store_true",
help="Enable auto-reload (not recommended for Windows)"
)
args = parser.parse_args()
# Handle reload default from environment
reload_default = os.getenv("RELOAD", "").lower() in ("true", "1", "yes")
if not args.reload:
args.reload = reload_default
return ServerConfig(
host=args.host,
port=args.port,
log_level=args.log_level,
reload=args.reload
)
def print_startup_info(host: str, port: int) -> None:
"""Print server startup information."""
print("Warbler CDA API Server")
print("=" * SEPARATOR_LENGTH)
print(f"App: {app.title}")
print(f"Host: {host}")
print(f"Port: {port}")
print()
print("Endpoints:")
print(f" Health check: http://{host}:{port}/health")
print(f" API docs: http://{host}:{port}/docs")
print()
print("Press Ctrl+C to stop")
def setup_logging(log_level: str) -> None:
"""Configure logging for both application and uvicorn."""
level = getattr(logging, log_level.upper())
# Configure application logging
logging.basicConfig(
level=level,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
# Configure uvicorn to use our logging
uvicorn_logger = logging.getLogger("uvicorn")
uvicorn_logger.setLevel(level)
def main() -> None:
"""Main entry point."""
try:
config = parse_args()
except ValueError as e:
print(f"Configuration error: {e}")
sys.exit(1)
setup_logging(config.log_level)
print_startup_info(config.host, config.port)
try:
uvicorn.run(
app,
host=config.host,
port=config.port,
log_level=config.log_level,
reload=config.reload,
)
except KeyboardInterrupt:
print("\nServer stopped by user")
sys.exit(0)
except ImportError as e:
print(f"Import Error: {e}")
traceback.print_exc()
sys.exit(1)
except OSError as e:
if "Address already in use" in str(e):
print(f"Port {config.port} is already in use")
else:
print(f"Network error: {e}")
sys.exit(1)
except Exception as e:
logger = logging.getLogger(__name__)
logger.error("Error starting server: %s", e, exc_info=True)
sys.exit(1)
if __name__ == "__main__":
main()
|