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()