Merge pull request #96 from socraticDevBlog/20250421-bootstrappingpostgresql

bootstraping posgresql allowing unit tests

Author: socraticDevBlog

backend/Pipfile

@@ -4,9 +4,12 @@ verify_ssl = true
 name = "pypi"
 
 [packages]
-fastapi = {extras = ["standard"], version = "*"}
 exceptiongroup = "*"
 tomli = "*"
+asyncpg = "*"
+python-dotenv = "*"
+fastapi = "*"
 
 [dev-packages]
 black = "*"
+pytest = "*"

backend/Pipfile.lock

@@ -53,3 +53,9 @@ pipenv run uvicorn app.main:app --reload
 ```
 
 <http://127.0.0.1:8000/docs>
+
+## unit tests
+
+```bash
+pipenv run pytest
+```

backend/README.md

@@ -0,0 +1,24 @@
+from dotenv import load_dotenv
+import os
+from app.scripts.postgresql import create_schema_if_not_exists, create_paste_table
+
+
+async def initialize_database():
+    """
+    Initialize the database by creating the schema and table if they don't exist.
+    """
+    load_dotenv()
+    conn_details = {
+        "host": os.getenv("DB_HOST", "localhost"),
+        "port": os.getenv("DB_PORT", "5432"),
+        "user": os.getenv("DB_USER"),
+        "password": os.getenv("DB_PASSWORD"),
+        "database": os.getenv("DB_NAME"),
+    }
+    schema_name = os.getenv("DB_SCHEMA", "public")
+
+    # Ensure schema and table exist
+    await create_schema_if_not_exists(
+        conn_details=conn_details, schema_name=schema_name
+    )
+    await create_paste_table(conn_details=conn_details, schema_name=schema_name)

backend/app/db_init.py

@@ -1,7 +1,22 @@
 from fastapi import FastAPI
 from app.routes import router
+from app.db_init import initialize_database
+from contextlib import asynccontextmanager
+import os
 
-app = FastAPI()
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """
+    Lifespan function to handle startup and shutdown events.
+    """
+    if os.getenv("INIT_DB", "true").lower() == "true":
+        await initialize_database()
+
+    yield  # Yield control back to FastAPI
+
+
+app = FastAPI(lifespan=lifespan)
 
 # Include the routes from the `routes.py` file
 app.include_router(router)

backend/app/main.py

@@ -0,0 +1,55 @@
+import asyncpg
+import os
+from dotenv import load_dotenv
+
+
+async def create_paste_table(conn_details, schema_name):
+    """
+    Create the 'Paste' table in the database if it doesn't exist.
+    """
+    conn = await asyncpg.connect(**conn_details)
+
+    create_table_query = f"""
+    CREATE TABLE IF NOT EXISTS {schema_name}.paste (
+        id TEXT PRIMARY KEY,
+        content TEXT NOT NULL,
+        client_id TEXT NOT NULL,
+        created_at BIGINT NOT NULL
+    );
+    """
+    await conn.execute(create_table_query)
+    print(
+        f"Table 'paste' ensured in database {conn_details['database']} under schema {schema_name}."
+    )
+
+    await conn.close()
+
+
+async def create_schema_if_not_exists(conn_details, schema_name):
+    conn = await asyncpg.connect(**conn_details)
+
+    query = f"CREATE SCHEMA IF NOT EXISTS {schema_name};"
+    await conn.execute(query)
+    print(
+        f"Schema '{schema_name}' ensured in database {conn_details['database']} under schema {schema_name}."
+    )
+
+    await conn.close()
+
+
+if __name__ == "__main__":
+    import asyncio
+
+    load_dotenv()
+    conn_details = {
+        "host": os.getenv("DB_HOST", "localhost"),
+        "port": os.getenv("DB_PORT", "5432"),
+        "user": os.getenv("DB_USER"),
+        "password": os.getenv("DB_PASSWORD"),
+        "database": os.getenv("DB_NAME"),
+    }
+    schema_name = os.getenv("DB_SCHEMA", "public")
+    asyncio.run(
+        create_schema_if_not_exists(conn_details=conn_details, schema_name=schema_name)
+    )
+    asyncio.run(create_paste_table(conn_details=conn_details, schema_name=schema_name))

backend/app/scripts/postgresql.py

@@ -0,0 +1,44 @@
+from unittest.mock import AsyncMock, patch
+from app.main import app
+
+from fastapi import Depends
+
+from fastapi.testclient import TestClient
+
+import os
+
+client = TestClient(app)  # Use FastAPI's TestClient for testing
+
+
+@patch("app.main.initialize_database", new_callable=AsyncMock)
+def test_app_initializes_database(mock_initialize_database):
+    """
+    Test that the app initializes the database when INIT_DB=true.
+    """
+    # Set the environment variable to simulate INIT_DB=true
+    os.environ["INIT_DB"] = "true"
+    client = TestClient(app)
+
+    with client:
+        pass
+
+    # Assert that initialize_database was called
+    mock_initialize_database.assert_called_once()
+
+
+@patch("app.main.initialize_database", new_callable=AsyncMock)
+def test_app_startup(mock_initialize_database):
+    # Test that the app starts without initializing the database
+    assert app  # Ensure the app instance is created
+    mock_initialize_database.assert_not_called()
+
+
+def get_db_connection():
+    # Return a real or mock database connection
+    pass
+
+
+@app.get("/example")
+async def example_route(db=Depends(get_db_connection)):
+    # Use the db connection
+    pass