🐛 fix(__main__.py): check if port is in use and get a free port if it is

 feat(__main__.py): add support for health check endpoint and use it to check server status
The `serve` function now checks if the specified port is in use and gets a free port if it is. This ensures that the server can run on a free port and avoids conflicts with other running applications. The `serve` function also now uses a health check endpoint to check the status of the server. The `setup_static_files` function now takes a `Path` object instead of a string for the static files directory. The `create_app` function now takes a `Path` object instead of a string for the static files directory.
This commit is contained in:
Gabriel Luiz Freitas Almeida 2023-06-16 17:39:52 -03:00
commit 319a57eb19
2 changed files with 39 additions and 4 deletions

View file

@ -6,7 +6,7 @@ from multiprocess import Process, cpu_count # type: ignore
import platform
from pathlib import Path
from typing import Optional
import socket
from rich.panel import Panel
from rich import box
from rich import print as rprint
@ -168,9 +168,13 @@ def serve(
)
webapp_process.start()
status_code = 0
# check if port is being used
if is_port_in_use(port, host):
port = get_free_port(port)
while status_code != 200:
try:
status_code = httpx.get(f"http://{host}:{port}").status_code
status_code = httpx.get(f"http://{host}:{port}/health").status_code
except Exception:
time.sleep(1)
@ -179,7 +183,7 @@ def serve(
webbrowser.open(f"http://{host}:{port}")
def setup_static_files(app: FastAPI, static_files_dir: str):
def setup_static_files(app: FastAPI, static_files_dir: Path):
"""
Setup the static files directory.
@ -194,6 +198,36 @@ def setup_static_files(app: FastAPI, static_files_dir: str):
)
def is_port_in_use(port, host="localhost"):
"""
Check if a port is in use.
Args:
port (int): The port number to check.
host (str): The host to check the port on. Defaults to 'localhost'.
Returns:
bool: True if the port is in use, False otherwise.
"""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex((host, port)) == 0
def get_free_port(port):
"""
Given a used port, find a free port.
Args:
port (int): The port number to check.
Returns:
int: A free port number.
"""
while is_port_in_use(port):
port += 1
return port
def print_banner(host, port):
# console = Console()

View file

@ -6,9 +6,10 @@ from fastapi.middleware.cors import CORSMiddleware
from langflow.api import router
from langflow.database.base import create_db_and_tables
from pathlib import Path
def create_app(static_path: str = "static"):
def create_app(static_path: Path = Path("src/frontend")):
"""Create the FastAPI app and include the router."""
app = FastAPI()