Refactor OpenAPI client script and backend server logic.
Simplify and streamline OpenAPI client generation and backend startup logic. Improved error handling, environment configuration, and self-signed SSL certificate management. Added support for generating OpenAPI schema via command-line argument.
This commit is contained in:
GotthardG
@ -1,4 +1,3 @@
|
||||
# app/main.py
|
||||
import os
|
||||
import json
|
||||
import tomllib
|
||||
@ -6,8 +5,6 @@ from pathlib import Path
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from app import ssl_heidi
|
||||
|
||||
|
||||
from app.routers import (
|
||||
address,
|
||||
contact,
|
||||
@ -25,7 +22,6 @@ from app.database import Base, engine, SessionLocal, load_sample_data, load_slot
|
||||
|
||||
# Utility function to fetch metadata from pyproject.toml
|
||||
def get_project_metadata():
|
||||
# Start from the current script's directory and search for pyproject.toml
|
||||
script_dir = Path(__file__).resolve().parent
|
||||
for parent in script_dir.parents:
|
||||
pyproject_path = parent / "pyproject.toml"
|
||||
@ -35,8 +31,6 @@ def get_project_metadata():
|
||||
name = pyproject["project"]["name"]
|
||||
version = pyproject["project"]["version"]
|
||||
return name, version
|
||||
|
||||
# If no pyproject.toml is found, raise FileNotFoundError
|
||||
raise FileNotFoundError(
|
||||
f"pyproject.toml not found in any parent directory of {script_dir}"
|
||||
)
|
||||
@ -45,21 +39,24 @@ def get_project_metadata():
|
||||
# Get project metadata from pyproject.toml
|
||||
project_name, project_version = get_project_metadata()
|
||||
app = FastAPI(
|
||||
title=project_name, # Syncs with project `name`
|
||||
title=project_name,
|
||||
description="Backend for next-gen sample management system",
|
||||
version=project_version, # Syncs with project `version`
|
||||
version=project_version,
|
||||
)
|
||||
|
||||
# Determine environment and configuration file path
|
||||
environment = os.getenv("ENVIRONMENT", "dev")
|
||||
config_file = Path(__file__).resolve().parent.parent / f"config_{environment}.json"
|
||||
|
||||
if not config_file.exists():
|
||||
raise FileNotFoundError(f"Config file '{config_file}' does not exist.")
|
||||
|
||||
# Load configuration
|
||||
with open(config_file) as f:
|
||||
config = json.load(f)
|
||||
|
||||
cert_path = config["ssl_cert_path"]
|
||||
key_path = config["ssl_key_path"]
|
||||
cert_path = config.get("ssl_cert_path", "ssl/cert.pem")
|
||||
key_path = config.get("ssl_key_path", "ssl/key.pem")
|
||||
|
||||
# Generate SSL Key and Certificate if not exist (only for development)
|
||||
if environment == "dev":
|
||||
@ -70,7 +67,7 @@ if environment == "dev":
|
||||
# Apply CORS middleware
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"], # Enable CORS for all origins
|
||||
allow_origins=["*"],
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
@ -118,60 +115,47 @@ app.include_router(spreadsheet.router, tags=["spreadsheet"])
|
||||
app.include_router(logistics.router, prefix="/logistics", tags=["logistics"])
|
||||
app.include_router(sample.router, prefix="/samples", tags=["samples"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
import uvicorn
|
||||
from pathlib import Path
|
||||
from app import ssl_heidi
|
||||
from dotenv import load_dotenv
|
||||
from multiprocessing import Process
|
||||
from time import sleep
|
||||
|
||||
# Load environment variables from .env file
|
||||
load_dotenv()
|
||||
|
||||
# Fetch environment values
|
||||
environment = os.getenv("ENVIRONMENT", "dev")
|
||||
port = int(os.getenv("PORT", 8000)) # Default to 8000 if PORT is not set
|
||||
is_ci = (
|
||||
os.getenv("CI", "false").lower() == "true"
|
||||
) # Detect if running in CI environment
|
||||
# Check if `generate-openapi` option is passed
|
||||
if len(sys.argv) > 1 and sys.argv[1] == "generate-openapi":
|
||||
from fastapi.openapi.utils import get_openapi
|
||||
|
||||
# Determine SSL certificate and key paths
|
||||
if environment == "prod":
|
||||
# Production environment must use proper SSL cert and key paths
|
||||
cert_path = os.getenv("VITE_SSL_CERT_PATH", "ssl/prod-cert.pem")
|
||||
key_path = os.getenv("VITE_SSL_KEY_PATH", "ssl/prod-key.pem")
|
||||
if not Path(cert_path).exists() or not Path(key_path).exists():
|
||||
raise FileNotFoundError(
|
||||
f"Production certificates not found."
|
||||
f"Make sure the following files exist:\n"
|
||||
f"Certificate: {cert_path}\nKey: {key_path}"
|
||||
)
|
||||
host = "0.0.0.0" # Allow external traffic
|
||||
print(
|
||||
f"Running in production mode with provided SSL certificates:\n"
|
||||
f" - Certificate: {cert_path}\n - Key: {key_path}"
|
||||
# Generate and save OpenAPI JSON file
|
||||
openapi_schema = get_openapi(
|
||||
title=app.title,
|
||||
version=app.version,
|
||||
description=app.description,
|
||||
routes=app.routes,
|
||||
)
|
||||
with open("openapi.json", "w") as f:
|
||||
json.dump(openapi_schema, f, indent=4)
|
||||
print("openapi.json generated successfully.")
|
||||
sys.exit(0) # Exit after generating the file
|
||||
|
||||
elif environment in ["test", "dev"]:
|
||||
# Test/Development environments use self-signed certificates
|
||||
# Default behavior: Run the server
|
||||
environment = os.getenv("ENVIRONMENT", "dev")
|
||||
port = int(os.getenv("PORT", 8000))
|
||||
is_ci = os.getenv("CI", "false").lower() == "true"
|
||||
|
||||
# Development or Test environment
|
||||
if environment in ["test", "dev"]:
|
||||
cert_path = "ssl/cert.pem"
|
||||
key_path = "ssl/key.pem"
|
||||
host = "127.0.0.1" # Restrict to localhost
|
||||
print(f"Running in {environment} mode with self-signed certificates...")
|
||||
|
||||
# Ensure self-signed certificates exist or generate them
|
||||
Path("ssl").mkdir(parents=True, exist_ok=True)
|
||||
if not Path(cert_path).exists() or not Path(key_path).exists():
|
||||
print(f"Generating self-signed SSL certificate at {cert_path}...")
|
||||
ssl_heidi.generate_self_signed_cert(cert_path, key_path)
|
||||
|
||||
host = "127.0.0.1"
|
||||
else:
|
||||
raise ValueError(
|
||||
f"Unknown environment: {environment}. "
|
||||
f"Must be one of 'prod', 'test', or 'dev'."
|
||||
)
|
||||
cert_path = os.getenv("VITE_SSL_CERT_PATH", "ssl/prod-cert.pem")
|
||||
key_path = os.getenv("VITE_SSL_KEY_PATH", "ssl/prod-key.pem")
|
||||
host = "0.0.0.0"
|
||||
|
||||
# Function to run the server
|
||||
def run_server():
|
||||
uvicorn.run(
|
||||
app,
|
||||
@ -182,18 +166,14 @@ if __name__ == "__main__":
|
||||
ssl_certfile=cert_path,
|
||||
)
|
||||
|
||||
# Continuous Integration handling
|
||||
# Run in CI mode
|
||||
if is_ci:
|
||||
from multiprocessing import Process
|
||||
from time import sleep
|
||||
|
||||
print("CI mode detected: Starting server in a subprocess...")
|
||||
server_process = Process(target=run_server)
|
||||
server_process.start()
|
||||
sleep(5) # Wait 5 seconds to ensure the server starts without errors
|
||||
server_process.terminate() # Terminate the server (test purposes)
|
||||
server_process.join() # Ensure proper cleanup
|
||||
sleep(5)
|
||||
server_process.terminate()
|
||||
server_process.join()
|
||||
print("CI: Server started and terminated successfully for test validation.")
|
||||
else:
|
||||
# Run the server normally
|
||||
run_server()
|
||||
|
||||
Reference in New Issue
Block a user