--- /dev/null
+# benchmark/run_benchmark.py
+import argparse
+import time
+import asyncio
+import httpx
+import requests
+from concurrent.futures import ThreadPoolExecutor
+
+# Server URLs (ensure these match your running servers)
+FLASK_URL = "http://127.0.0.1:3000/"
+FASTAPI_URL = "http://127.0.0.1:8000/"
+NUM_REQUESTS = 100
+
+def fetch_url_sync(url):
+ try:
+ response = requests.get(url, timeout=10) # Increased timeout for potentially slow server
+ response.raise_for_status() # Raise an exception for bad status codes
+ return response.status_code
+ except requests.exceptions.RequestException as e:
+ print(f"Request to {url} failed: {e}")
+ return None
+
+def run_flask_benchmark():
+ print(f"Starting Flask benchmark: {NUM_REQUESTS} requests to {FLASK_URL}...")
+ start_time = time.perf_counter()
+
+ with ThreadPoolExecutor(max_workers=NUM_REQUESTS) as executor:
+ futures = [executor.submit(fetch_url_sync, FLASK_URL) for _ in range(NUM_REQUESTS)]
+ results = [future.result() for future in futures]
+
+ end_time = time.perf_counter()
+ total_time = end_time - start_time
+ successful_requests = sum(1 for r in results if r == 200)
+ print(f"Flask benchmark: {successful_requests}/{NUM_REQUESTS} successful requests in {total_time:.2f} seconds.")
+ return total_time
+
+async def fetch_url_async(client, url):
+ try:
+ response = await client.get(url, timeout=10) # Increased timeout
+ response.raise_for_status()
+ return response.status_code
+ except httpx.RequestError as e:
+ print(f"Request to {url} failed: {e}")
+ return None
+
+async def run_fastapi_benchmark_async():
+ print(f"Starting FastAPI benchmark: {NUM_REQUESTS} requests to {FASTAPI_URL}...")
+ start_time = time.perf_counter()
+
+ async with httpx.AsyncClient() as client:
+ tasks = [fetch_url_async(client, FASTAPI_URL) for _ in range(NUM_REQUESTS)]
+ results = await asyncio.gather(*tasks)
+
+ end_time = time.perf_counter()
+ total_time = end_time - start_time
+ successful_requests = sum(1 for r in results if r == 200)
+ print(f"FastAPI benchmark: {successful_requests}/{NUM_REQUESTS} successful requests in {total_time:.2f} seconds.")
+ return total_time
+
+def run_fastapi_benchmark():
+ return asyncio.run(run_fastapi_benchmark_async())
+
+if __name__ == "__main__":
+ parser = argparse.ArgumentParser(description="Run web server benchmarks.")
+ parser.add_argument(
+ "framework",
+ choices=["flask", "fastapi"],
+ help="Specify the framework to benchmark (flask or fastapi)"
+ )
+ args = parser.parse_args()
+
+ if args.framework == "flask":
+ run_flask_benchmark()
+ elif args.framework == "fastapi":
+ run_fastapi_benchmark()
+ else:
+ print("Invalid framework specified. Choose 'flask' or 'fastapi'.")
\ No newline at end of file
# tests/test_fastapi_route.py
import httpx
-import subprocess, asyncio, os, signal, time # signal and time may not be needed if Popen is handled well
+import asyncio
import pytest # For @pytest.mark.asyncio
+import uvicorn
+import threading
+import time
+from multiprocessing import Process # Using Process for better isolation
+
+# Server configuration
+HOST = "127.0.0.1"
+PORT = 8000
+
+class UvicornServer(threading.Thread):
+ def __init__(self, app_module_str):
+ super().__init__(daemon=True)
+ self.app_module_str = app_module_str
+ self.server_started = threading.Event()
+ self.config = uvicorn.Config(app_module_str, host=HOST, port=PORT, log_level="info")
+ self.server = uvicorn.Server(config=self.config)
+
+ def run(self):
+ # Need to set a new event loop for the thread if running asyncio components
+ # For uvicorn.Server.serve(), it typically manages its own loop or integrates
+ # with an existing one if started from an async context.
+ # However, running it in a separate thread needs care.
+ # Simpler: uvicorn.run() which handles loop creation.
+ uvicorn.run(self.app_module_str, host=HOST, port=PORT, log_level="warning") # log_level warning to reduce noise
+
+ # UvicornServer using Process for cleaner start/stop
+ # This might be more robust for test isolation.
+
+def run_server_process(app_module_str, host, port):
+ uvicorn.run(app_module_str, host=host, port=port, log_level="warning")
-# Adjusted to be an async function and use uvicorn
async def start_server_fastapi():
- # Ensure CWD is project root
- project_root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
- # Command to run FastAPI app with uvicorn
- # Assuming app_fastapi/app.py and app instance is named 'app'
- cmd = ["uvicorn", "app_fastapi.app:app", "--host", "0.0.0.0", "--port", "8000"]
-
- proc = subprocess.Popen(
- cmd,
- cwd=project_root,
- stdout=subprocess.PIPE,
- stderr=subprocess.PIPE
- )
- await asyncio.sleep(1.5) # Allow uvicorn to start
- return proc
+ # Using Process to run Uvicorn. This provides better isolation and cleanup.
+ proc = Process(target=run_server_process, args=("app_fastapi.app:app", HOST, PORT), daemon=True)
+ proc.start()
+ await asyncio.sleep(2.0) # Increased sleep to ensure server is fully up
+ if not proc.is_alive():
+ raise RuntimeError("FastAPI server process failed to start.")
+ return proc # Return the process object
@pytest.mark.asyncio
async def test_home_returns_html_fastapi():
- proc = await start_server_fastapi()
+ server_process = await start_server_fastapi()
try:
- async with httpx.AsyncClient() as client:
- r = await client.get("http://127.0.0.1:8000/") # Default FastAPI port
+ async with httpx.AsyncClient(timeout=10) as client:
+ r = await client.get(f"http://{HOST}:{PORT}/")
assert r.status_code == 200
- assert "<h1>Slow FastAPI Demo</h1>" in r.text # Expected content
+ assert "<h1>FastAPI Server: 3s Artificial Delay Demo</h1>" in r.text # Expected content
finally:
- proc.terminate()
- try:
- # Use communicate to get output and ensure process is reaped
- stdout, stderr = proc.communicate(timeout=5)
- # print(f"FastAPI stdout:\n{stdout.decode(errors='replace')}") # Optional: for debugging
- # print(f"FastAPI stderr:\n{stderr.decode(errors='replace')}") # Optional: for debugging
- except subprocess.TimeoutExpired:
- print("FastAPI server did not terminate/communicate gracefully, killing.")
- proc.kill()
- proc.wait() # Ensure kill completes
\ No newline at end of file
+ if server_process and server_process.is_alive():
+ server_process.terminate() # Send SIGTERM
+ server_process.join(timeout=5) # Wait for termination
+ if server_process.is_alive(): # Still alive after timeout
+ print("FastAPI server process did not terminate gracefully, killing.")
+ server_process.kill() # Force kill
+ server_process.join(timeout=5) # Wait for kill
+ # print("FastAPI server process stopped.") # Optional debug
\ No newline at end of file
projects / fastapi-vs-flask / .git / commitdiff