86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
import logging
|
|
import socket
|
|
import sys
|
|
|
|
import uvicorn
|
|
from fastapi import Depends, FastAPI
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.api.api import api_router
|
|
from app.core.config import settings
|
|
from app.db.session import get_db
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = FastAPI(
|
|
title=settings.PROJECT_NAME,
|
|
openapi_url=f"{settings.API_V1_STR}/openapi.json",
|
|
description="Todo application API",
|
|
version="0.1.0",
|
|
)
|
|
|
|
@app.get("/health", tags=["health"], summary="Health check endpoint")
|
|
def health_check(db: Session = Depends(get_db)):
|
|
"""
|
|
Check the health of the application:
|
|
|
|
- **status**: Overall application status
|
|
- **database**: Database connection status
|
|
"""
|
|
try:
|
|
# Make a simple query to check if the database is responsive
|
|
db.execute("SELECT 1")
|
|
db_status = "healthy"
|
|
except Exception:
|
|
db_status = "unhealthy"
|
|
|
|
return {
|
|
"status": "ok",
|
|
"database": db_status
|
|
}
|
|
|
|
app.include_router(api_router, prefix=settings.API_V1_STR)
|
|
|
|
def find_available_port(start_port, max_attempts=10):
|
|
"""Try to find an available port starting from start_port"""
|
|
for port in range(start_port, start_port + max_attempts):
|
|
try:
|
|
# Try to create a socket with the port
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind((settings.HOST, port))
|
|
# If we get here, the port is available
|
|
return port
|
|
except OSError:
|
|
# Port is already in use, try the next one
|
|
continue
|
|
|
|
# If we get here, no ports were available in our range
|
|
port_range_end = start_port + max_attempts - 1
|
|
logger.warning(
|
|
f"Could not find an available port in range {start_port}-{port_range_end}"
|
|
)
|
|
return None
|
|
|
|
if __name__ == "__main__":
|
|
port = settings.PORT
|
|
|
|
# Check if the port is already in use
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
|
s.bind((settings.HOST, port))
|
|
except OSError:
|
|
logger.warning(
|
|
f"Port {port} is already in use. Trying to find an available port..."
|
|
)
|
|
port = find_available_port(port + 1)
|
|
|
|
if port is None:
|
|
logger.error("Could not find an available port. Exiting.")
|
|
sys.exit(1)
|
|
|
|
logger.info(f"Using alternative port: {port}")
|
|
|
|
# Start the server with the available port
|
|
uvicorn.run("main:app", host=settings.HOST, port=port, reload=True) |