WIP

Author: jason810496

.github/workflows/codecov.yml

@@ -33,7 +33,7 @@ jobs:
             curl -LsSf https://astral.sh/uv/install.sh | sh
             echo "$HOME/.local/bin" >> $GITHUB_PATH
         - name: Install dependencies
-          run: uv sync --extra dev
+          run: uv sync --group postgresql-drivers --group test
         - name: Start PostgreSQL
           run: |
             cp pgmq_postgres.template.env pgmq_postgres.env

.github/workflows/examples.yml

@@ -44,9 +44,7 @@ jobs:
       
       - name: Install dependencies
         run: |
-          uv sync --extra dev
-          # Install additional dependencies for examples
-          uv pip install fastapi uvicorn httpx
+          uv sync --all-groups --no-group docs
       
       - name: Start PostgreSQL
         run: |

examples/fastapi_pub_sub/pyproject.toml

@@ -2,6 +2,8 @@
 import os
 import pytest
 
+import logging
+
 
 def pytest_addoption(parser):
     """Add custom command-line options for pytest."""
@@ -12,6 +14,23 @@ def pytest_addoption(parser):
         help="Specify the database name to use for testing",
     )
 
+@pytest.fixture(scope="module")
+def configure_logger():
+    logging.basicConfig(
+        level=logging.INFO,
+        format="[%(levelname)s][%(asctime)s][%(name)s] %(message)s"
+    )
+
+
+@pytest.fixture(scope="module")
+def examples_dir():
+    """Return the path to the examples directory."""
+    return os.path.join(
+        os.path.dirname(os.path.dirname(__file__)),
+        "examples",
+        "fastapi_pub_sub"
+    )
+
 
 @pytest.fixture(scope="module")
 def database_url(request):

examples/fastapi_pub_sub/requirements.txt

@@ -1,26 +1,20 @@
 """Integration tests for FastAPI pub/sub example with subprocess."""
-import asyncio
+
 import os
 import subprocess
 import sys
 import time
 import signal
+import logging
+
 import pytest
-import requests
-from sqlalchemy import create_engine, text
+import httpx
+from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 
 from pgmq_sqlalchemy import op
 
-
-@pytest.fixture(scope="module")
-def examples_dir():
-    """Return the path to the examples directory."""
-    return os.path.join(
-        os.path.dirname(os.path.dirname(os.path.dirname(__file__))),
-        "examples",
-        "fastapi_pub_sub"
-    )
+logger = logging.getLogger(__name__)
 
 
 @pytest.fixture(scope="module")
@@ -29,49 +23,35 @@ def test_queue_name():
     return "test_integration_order_queue"
 
 
-@pytest.fixture(scope="module")
-def database_url(request):
-    """Get database URL from environment or CLI."""
-    db_name = request.config.getoption("--db-name")
-    if not db_name:
-        db_name = os.getenv("SQLALCHEMY_DB", "postgres")
-    
-    host = os.getenv("SQLALCHEMY_HOST", "localhost")
-    port = os.getenv("SQLALCHEMY_PORT", "5432")
-    user = os.getenv("SQLALCHEMY_USER", "postgres")
-    password = os.getenv("SQLALCHEMY_PASSWORD", "postgres")
-    
-    return f"postgresql+psycopg2://{user}:{password}@{host}:{port}/{db_name}"
-
-
 @pytest.fixture(scope="module", autouse=True)
-def api_instance(examples_dir, database_url, test_queue_name):
+def api_instance(examples_dir, sync_database_url, test_queue_name):
     """Fixture to spin up the API server as a subprocess."""
     # Update the API to use test queue
     api_py = os.path.join(examples_dir, "api.py")
 
     # Set environment variables for the subprocess
     env = os.environ.copy()
-    env["DATABASE_URL"] = database_url
+    env["DATABASE_URL"] = sync_database_url
     env["QUEUE_NAME"] = test_queue_name
 
     # Start the API server
     process = subprocess.Popen(
         [sys.executable, api_py],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
+        stdout=sys.stdout,
+        stderr=sys.stderr,
         env=env,
         preexec_fn=os.setsid if hasattr(os, 'setsid') else None
     )
+    logger.info("Create API Server Process")
 
     # Wait for the server to start
     max_attempts = 30
     for i in range(max_attempts):
         try:
-            response = requests.get("http://localhost:8000/health", timeout=1)
+            response = httpx.get("http://localhost:8000/health", timeout=1)
             if response.status_code == 200:
                 break
-        except requests.exceptions.RequestException:
+        except Exception:
             time.sleep(1)
     else:
         # Kill the process if it didn't start
@@ -80,7 +60,8 @@ def api_instance(examples_dir, database_url, test_queue_name):
         else:
             process.terminate()
         pytest.fail("API server failed to start")
-    
+
+    logger.info("API Server is healthy")
     yield process
 
     # Teardown: kill the API server
@@ -89,30 +70,32 @@ def api_instance(examples_dir, database_url, test_queue_name):
     else:
         process.terminate()
     process.wait(timeout=10)
+    logger.info("Terminate API Server")
 
 
 @pytest.fixture(scope="module", autouse=True)
-def consumer_instance(examples_dir, database_url, test_queue_name, api_instance):
+def consumer_instance(examples_dir, async_database_url, test_queue_name):
     """Fixture to spin up the consumer as a subprocess."""
     # Update the consumer to use test queue
     consumer_py = os.path.join(examples_dir, "consumer.py")
 
     # Set environment variables for the subprocess
     env = os.environ.copy()
-    env["DATABASE_URL"] = database_url
+    env["DATABASE_URL"] = async_database_url
     env["QUEUE_NAME"] = test_queue_name
 
     # Start the consumer
     process = subprocess.Popen(
         [sys.executable, consumer_py],
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
+        stdout=sys.stdout,
+        stderr=sys.stderr,
         env=env,
         preexec_fn=os.setsid if hasattr(os, 'setsid') else None
     )
 
     # Give the consumer some time to start
     time.sleep(3)
+    logger.info("Create Consumer Process")
 
     yield process
 
@@ -122,24 +105,25 @@ def consumer_instance(examples_dir, database_url, test_queue_name, api_instance)
     else:
         process.terminate()
     process.wait(timeout=10)
+    logger.info("Terminate Consumer Process")
 
 
-def test_api_consumer_integration(api_instance, consumer_instance, database_url):
+def test_api_consumer_integration(sync_database_url):
     """Test creating 100 orders parallelly and waiting for consumer to process them all."""
     import concurrent.futures
 
     # Create 100 orders in parallel
     num_orders = 100
 
-    def create_order(order_num):
+    def create_order(order_num: int):
         """Helper function to create a single order."""
         order_data = {
             "customer_name": f"Customer {order_num}",
             "product_name": f"Product {order_num}",
             "quantity": order_num % 10 + 1,
             "price": 10.0 + (order_num % 50)
         }
-        response = requests.post("http://localhost:8000/orders", json=order_data, timeout=5)
+        response = httpx.post("http://localhost:8000/orders", json=order_data, timeout=5)
         return response.status_code == 201, response.json() if response.status_code == 201 else None
 
     # Create orders in parallel
@@ -150,15 +134,17 @@ def create_order(order_num):
     # Check that all orders were created successfully
     successful_orders = sum(1 for success, _ in results if success)
     assert successful_orders == num_orders, f"Only {successful_orders}/{num_orders} orders were created"
+    logger.info("Create %d successful orders via API Server", successful_orders)
 
     # Wait for the consumer to process all messages
     # Check the queue periodically until it's empty
-    engine = create_engine(database_url)
+    engine = create_engine(sync_database_url)
     SessionLocal = sessionmaker(bind=engine)
 
     max_wait = 120  # Wait up to 2 minutes
     start_time = time.time()
 
+    logger.info("Wait for Consumer to process all the orders");
     while time.time() - start_time < max_wait:
         # Check queue metrics to see if there are any messages left
         with SessionLocal() as session:
@@ -167,15 +153,20 @@ def create_order(order_num):
 
             try:
                 metrics = op.metrics(test_queue, session=session, commit=True)
-                if metrics.queue_length == 0:
-                    # All messages have been processed
-                    break
+                if metrics:
+                    logger.info("%s queue metrics: %s", test_queue, str(metrics))
+                    if metrics.queue_length == 0:
+                        # All messages have been processed
+                        break
             except Exception as e:
                 # Queue might not exist yet or other error
                 print(f"Error checking metrics: {e}")
 
         time.sleep(2)
     else:
+        metrics = op.metrics(test_queue, session=session, commit=True)
+        if metrics:
+            logger.info("%s queue metrics: %s", test_queue, str(metrics))
         pytest.fail(f"Consumer did not process all messages within {max_wait} seconds")
 
     # Verify that all messages were processed

examples_tests/conftest.py

@@ -38,28 +38,46 @@ pg8000 = ["pg8000>=1.31.2"]
 psycopg = ["psycopg>=3.2.1"]
 psycopg2-binary = ["psycopg2-binary>=2.9.9"]
 psycopg2cffi = ["psycopg2cffi>=2.9.0"]
+
+# Include dependencies by `uv sync --group dev`
+[dependency-groups]
 dev = [
-    # postgresql drivers
+    {include-group = "postgresql-drivers"},
+    {include-group = "test"},
+    {include-group = "docs"},
+    {include-group = "fastapi-pub-sub-example"},
+    "rich>=14.2.0",
+    "ruff>=0.14.10",
+    "libcst>=1.8.6",
+]
+postgresql-drivers = [
     "asyncpg>=0.29.0",
     "greenlet>=3.0.3",
     "pg8000>=1.31.2",
     "psycopg>=3.2.1",
     "psycopg2-binary>=2.9.9",
     "psycopg2cffi>=2.9.0",
-    # testing
+]
+test = [
     "pytest>=7.4.4,<8.0",
     "pytest-asyncio>=0.23.8",
     "pytest-lazy-fixture>=0.6.3",
     "pytest-cov>=5.0.0",
     "pytest-xdist>=3.6.1",
     "filelock>=3.15.4",
-    # docs
+]
+docs = [
     "sphinx>=7.3.7",
     "sphinx-autobuild>=2024.4.16",
     "sphinx-rtd-theme>=2.0.0",
     "sphinx-copybutton>=0.5.2",
 ]
+fastapi-pub-sub-example = [
+    "fastapi>=0.104.0",
+    "httpx>=0.28.1",
+    "uvicorn>=0.24.0",
+]
 
 [build-system]
 requires = ["hatchling"]
-build-backend = "hatchling.build"
+build-backend = "hatchling.build"