47 - Advanced PostgreSQL Features

from typing import Dict, Any
import psycopg2
from psycopg2.extras import Json

# Connect to tasks_db PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="tasks_db", user="postgres", password="password", host="localhost"
)
# Create a cursor for executing queries
cursor: psycopg2.cursor = conn.cursor()

# Add JSONB column to store metadata if it doesn't exist
cursor.execute("ALTER TABLE tasks ADD COLUMN IF NOT EXISTS metadata JSONB;")
# Create GIN index for efficient JSONB queries
cursor.execute("CREATE INDEX IF NOT EXISTS metadata_idx ON tasks USING GIN (metadata);")
# Commit changes to the database
conn.commit()

# Define sample metadata for a task
task_metadata: Dict[str, Any] = {"priority": "high", "owner": "finance_team"}
# Update the task with JSONB metadata for task_id T001
cursor.execute(
    "UPDATE tasks SET metadata = %s WHERE task_id = %s;",
    (Json(task_metadata), "T001")  # Json adapter converts dict to JSONB
)
# Commit the update
conn.commit()

# Query tasks where owner is finance_team, extracting priority
cursor.execute(
    "SELECT task_id, metadata->'priority' AS priority FROM tasks WHERE metadata->>'owner' = %s;",
    ("finance_team",)  # Use ->> to get text value of owner
)
# Fetch all query results
results: list = cursor.fetchall()
# Print results for debugging
print("JSONB Query Results:", results)

# Expected Output:
# JSONB Query Results: [('T001', 'high')]

# Close the cursor
cursor.close()
# Close the database connection
conn.close()

from typing import List, Tuple
import psycopg2

# Connect to tasks_db PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="tasks_db", user="postgres", password="password", host="localhost"
)
# Create a cursor for executing queries
cursor: psycopg2.cursor = conn.cursor()

# Add tsvector column for full-text search if it doesn't exist
cursor.execute("ALTER TABLE tasks ADD COLUMN IF NOT EXISTS search_vector TSVECTOR;")
# Update tasks to generate tsvector from description using English dictionary
cursor.execute(
    "UPDATE tasks SET search_vector = to_tsvector('english', description);"
)
# Create GIN index for efficient full-text search queries
cursor.execute("CREATE INDEX IF NOT EXISTS search_idx ON tasks USING GIN(search_vector);")
# Commit changes to the database
conn.commit()

# Search tasks with descriptions containing "inventory"
cursor.execute(
    "SELECT task_id, description FROM tasks WHERE search_vector @@ to_tsquery('english', %s);",
    ("inventory",)  # to_tsquery creates a query for the word "inventory"
)
# Fetch all search results
results: List[Tuple[str, str]] = cursor.fetchall()
# Print results for debugging
print("Full-Text Search Results:", results)

# Expected Output:
# Full-Text Search Results: [('T002', 'Validate Halal Mouse inventory')]

# Close the cursor
cursor.close()
# Close the database connection
conn.close()

from typing import List, Tuple
import psycopg2

# Connect to tasks_db PostgreSQL database
conn: psycopg2.connection = psycopg2.connect(
    dbname="tasks_db", user="postgres", password="password", host="localhost"
)
# Create a cursor for executing queries
cursor: psycopg2.cursor = conn.cursor()

# Create materialized view to count tasks by status
cursor.execute("""
CREATE MATERIALIZED VIEW IF NOT EXISTS task_status_summary AS
SELECT status, COUNT(*) AS task_count
FROM tasks
GROUP BY status;
""")
# Commit the view creation
conn.commit()

# Refresh materialized view to update data
cursor.execute("REFRESH MATERIALIZED VIEW task_status_summary;")
# Commit the refresh
conn.commit()

# Query the materialized view to retrieve status counts
cursor.execute("SELECT status, task_count FROM task_status_summary;")
# Fetch all query results
results: List[Tuple[str, int]] = cursor.fetchall()
# Print results for debugging
print("Materialized View Results:", results)

# Expected Output:
# Materialized View Results: [('Completed', 1), ('Pending', 1), ('In Progress', 1)]

# Close the cursor
cursor.close()
# Close the database connection
conn.close()

# File: de-onboarding/create_tasks_db.py
from typing import List, Tuple
import psycopg2

def create_tasks_db(config: dict) -> None:
    """Create and seed tasks_db in PostgreSQL."""
    # Connect to default postgres database to create tasks_db
    conn: psycopg2.connection = psycopg2.connect(
        dbname="postgres", user=config["user"], password=config["password"], host=config["host"]
    )
    # Enable autocommit for database creation
    conn.set_session(autocommit=True)
    # Create a cursor for executing queries
    cursor: psycopg2.cursor = conn.cursor()

    # Create tasks_db if it doesn't exist
    cursor.execute("CREATE DATABASE tasks_db;")
    # Close the cursor and connection
    cursor.close()
    conn.close()

    # Connect to the new tasks_db
    conn = psycopg2.connect(**config)
    cursor = conn.cursor()

    # Create tasks table with task_id as primary key
    cursor.execute("""
    CREATE TABLE IF NOT EXISTS tasks (
        task_id TEXT PRIMARY KEY,
        description TEXT,
        status TEXT
    );
    """)

    # Define sample tasks for seeding
    tasks: List[Tuple[str, str, str]] = [
        ("T001", "Process Halal Laptop sales", "Completed"),
        ("T002", "Validate Halal Mouse inventory", "Pending"),
        ("T003", "Update Halal Keyboard pricing", "In Progress")
    ]
    # Insert tasks, ignoring duplicates
    cursor.executemany(
        "INSERT INTO tasks (task_id, description, status) VALUES (%s, %s, %s) ON CONFLICT DO NOTHING;",
        tasks
    )

    # Commit changes to the database
    conn.commit()
    # Close the cursor and connection
    cursor.close()
    conn.close()
    # Confirm database creation
    print("Created and seeded tasks_db")

min_price: 10.0
max_quantity: 100
required_fields:
  - product
  - price
  - quantity
product_prefix: 'Halal'
max_decimals: 2
database:
  dbname: tasks_db
  user: postgres
  password: password
  host: localhost

# File: de-onboarding/utils.py
from typing import Any, Dict
import yaml

def read_config(config_path: str) -> Dict[str, Any]:
    """Read YAML configuration file."""
    # Print the config file path for debugging
    print(f"Opening config: {config_path}")
    # Open and read the YAML file
    with open(config_path, "r") as file:
        # Parse YAML content into a dictionary
        config = yaml.safe_load(file)
    # Print the loaded config for debugging
    print(f"Loaded config: {config}")
    # Return the config dictionary
    return config

# File: de-onboarding/task_processor.py
from typing import Dict, Any, List, Tuple, Optional
import psycopg2
from psycopg2.extras import Json
from pydantic import BaseModel
import utils

# Pydantic model for JSONB metadata validation
class TaskMetadata(BaseModel):
    priority: Optional[str] = None  # Task priority, nullable
    owner: Optional[str] = None     # Task owner, nullable
    notes: Optional[str] = None     # Additional notes, nullable

# Pydantic model for full-text search results
class TaskSearchResult(BaseModel):
    task_id: str        # Unique task identifier
    description: str   # Task description

# Pydantic model for materialized view summaries
class TaskSummary(BaseModel):
    status: str        # Task status
    task_count: int    # Number of tasks with this status

# Class to manage task database operations
class TaskProcessor:
    def __init__(self, config: Dict[str, Any]):
        """Initialize database connection."""
        # Store database configuration
        self.config = config["database"]
        # Connect to tasks_db using config
        self.conn: psycopg2.connection = psycopg2.connect(**self.config)
        # Create a cursor for executing queries
        self.cursor: psycopg2.cursor = self.conn.cursor()
        # Confirm connection
        print("Connected to tasks_db")

    def setup_database(self) -> None:
        """Add JSONB, tsvector, and materialized view to tasks table."""
        # Add JSONB column for metadata if it doesn't exist
        self.cursor.execute("ALTER TABLE tasks ADD COLUMN IF NOT EXISTS metadata JSONB;")
        # Create GIN index for efficient JSONB queries
        self.cursor.execute("CREATE INDEX IF NOT EXISTS metadata_idx ON tasks USING GIN (metadata);")
        # Add tsvector column for full-text search if it doesn't exist
        self.cursor.execute("ALTER TABLE tasks ADD COLUMN IF NOT EXISTS search_vector TSVECTOR;")
        # Generate tsvector from descriptions using English dictionary
        self.cursor.execute(
            "UPDATE tasks SET search_vector = to_tsvector('english', description);"
        )
        # Create GIN index for efficient full-text search
        self.cursor.execute("CREATE INDEX IF NOT EXISTS search_idx ON tasks USING GIN(search_vector);")
        # Create materialized view to summarize task counts by status
        self.cursor.execute("""
        CREATE MATERIALIZED VIEW IF NOT EXISTS task_status_summary AS
        SELECT status, COUNT(*) AS task_count
        FROM tasks
        GROUP BY status;
        """)
        # Commit changes to the database
        self.conn.commit()
        # Confirm setup completion
        print("Database setup completed")

    def insert_metadata(self, task_id: str, metadata: Dict[str, Any]) -> None:
        """Insert JSONB metadata for a task."""
        # Update task with JSONB metadata
        self.cursor.execute(
            "UPDATE tasks SET metadata = %s WHERE task_id = %s;",
            (Json(metadata), task_id)  # Json adapter converts dict to JSONB
        )
        # Commit the update
        self.conn.commit()
        # Log the insertion
        print(f"Inserted metadata for task {task_id}")

    def query_metadata(self, owner: str) -> List[TaskMetadata]:
        """Query tasks by JSONB owner field."""
        # Query tasks where owner matches the input
        self.cursor.execute(
            "SELECT metadata FROM tasks WHERE metadata->>'owner' = %s;",
            (owner,)
        )
        # Initialize results list
        results = []
        # Process each row
        for row in self.cursor.fetchall():
            # Handle null JSONB by using empty dict
            metadata = row[0] or {}
            # Print raw metadata for debugging
            print(f"Raw metadata: {metadata}")
            # Validate metadata with Pydantic
            results.append(TaskMetadata(**metadata))
        # Print results for debugging
        print(f"Metadata query results: {results}")
        # Return validated results
        return results

    def search_tasks(self, query: str) -> List[TaskSearchResult]:
        """Search tasks by description using full-text search."""
        # Search tasks matching the query
        self.cursor.execute(
            "SELECT task_id, description FROM tasks WHERE search_vector @@ to_tsquery('english', %s);",
            (query,)  # to_tsquery creates a query for the input word
        )
        # Convert results to Pydantic models
        results = [TaskSearchResult(task_id=row[0], description=row[1]) for row in self.cursor.fetchall()]
        # Print results for debugging
        print(f"Search results: {results}")
        # Return search results
        return results

    def refresh_summary(self) -> None:
        """Refresh materialized view to update task status counts."""
        # Refresh the materialized view
        self.cursor.execute("REFRESH MATERIALIZED VIEW task_status_summary;")
        # Commit the refresh
        self.conn.commit()
        # Confirm refresh
        print("Materialized view refreshed")

    def query_summary(self) -> List[TaskSummary]:
        """Query task status summary from materialized view."""
        # Query status counts from the materialized view
        self.cursor.execute("SELECT status, task_count FROM task_status_summary;")
        # Convert results to Pydantic models
        results = [TaskSummary(status=row[0], task_count=row[1]) for row in self.cursor.fetchall()]
        # Print results for debugging
        print(f"Summary results: {results}")
        # Return summary results
        return results

    def close(self) -> None:
        """Close database connection."""
        # Close the cursor
        self.cursor.close()
        # Close the connection
        self.conn.close()
        # Confirm closure
        print("Database connection closed")

def main() -> None:
    """Main function to enhance task database."""
    # Define path to config file
    config_path = "data/config.yaml"
    # Read configuration
    config = utils.read_config(config_path)

    # Initialize task processor
    processor = TaskProcessor(config)
    # Set up database with advanced features
    processor.setup_database()

    # Insert sample metadata for task T001
    processor.insert_metadata("T001", {"priority": "high", "owner": "finance_team", "notes": "urgent"})

    # Query metadata for finance_team owner
    metadata_results = processor.query_metadata("finance_team")

    # Search tasks for "inventory"
    search_results = processor.search_tasks("inventory")

    # Refresh and query task status summary
    processor.refresh_summary()
    summary_results = processor.query_summary()

    # Close the database connection
    processor.close()

    # Print final report
    print("\nTask Database Report:")
    print(f"Metadata Results: {metadata_results}")
    print(f"Search Results: {search_results}")
    print(f"Summary Results: {summary_results}")

if __name__ == "__main__":
    # Run the main function
    main()

# File: de-onboarding/test_task_processor.py
import pytest
from task_processor import TaskProcessor, TaskMetadata, TaskSearchResult, TaskSummary
import utils
from psycopg2.extras import Json

@pytest.fixture
def processor():
    """Fixture to set up and tear down TaskProcessor."""
    # Read configuration
    config = utils.read_config("data/config.yaml")
    # Initialize TaskProcessor
    proc = TaskProcessor(config)
    # Set up database
    proc.setup_database()
    # Yield processor for tests
    yield proc
    # Close connection after tests
    proc.close()

def test_insert_and_query_metadata(processor):
    """Test inserting and querying JSONB metadata."""
    # Insert metadata for task T001
    processor.insert_metadata("T001", {"priority": "high", "owner": "finance_team"})
    # Query metadata for finance_team
    results = processor.query_metadata("finance_team")
    # Verify one result
    assert len(results) == 1
    # Check priority and owner
    assert results[0].priority == "high"
    assert results[0].owner == "finance_team"

def test_search_tasks(processor):
    """Test full-text search functionality."""
    # Search for tasks with "inventory"
    results = processor.search_tasks("inventory")
    # Verify one result
    assert len(results) == 1
    # Check task_id and description
    assert results[0].task_id == "T002"
    assert results[0].description == "Validate Halal Mouse inventory"

def test_empty_search(processor):
    """Test full-text search with no results."""
    # Search for nonexistent term
    results = processor.search_tasks("nonexistent")
    # Verify no results
    assert len(results) == 0

def test_summary(processor):
    """Test materialized view summary."""
    # Refresh materialized view
    processor.refresh_summary()
    # Query summary
    results = processor.query_summary()
    # Verify three status counts
    assert len(results) == 3
    # Check each status has one task
    assert any(r.status == "Completed" and r.task_count == 1 for r in results)
    assert any(r.status == "Pending" and r.task_count == 1 for r in results)
    assert any(r.status == "In Progress" and r.task_count == 1 for r in results)

def test_invalid_metadata(processor):
    """Test handling invalid JSONB metadata."""
    # Insert metadata with null priority
    processor.insert_metadata("T002", {"priority": None, "owner": "sales_team"})
    # Query metadata for sales_team
    results = processor.query_metadata("sales_team")
    # Verify one result
    assert len(results) == 1
    # Check null priority and owner
    assert results[0].priority is None
    assert results[0].owner == "sales_team"

def test_large_metadata(processor):
    """Test handling large JSONB metadata."""
    # Insert metadata with multiple fields
    processor.insert_metadata("T003", {"priority": "low", "owner": "sales_team", "notes": "urgent"})
    # Query metadata for sales_team
    results = processor.query_metadata("sales_team")
    # Verify one result
    assert len(results) == 1
    # Check all fields
    assert results[0].priority == "low"
    assert results[0].owner == "sales_team"
    assert results[0].notes == "urgent"

def test_invalid_jsonb_input(processor):
    """Test handling invalid JSONB input."""
    # Insert valid metadata to ensure task exists
    processor.insert_metadata("T004", {"priority": "medium", "owner": "ops_team"})
    # Attempt to insert invalid JSONB data (non-dict)
    invalid_data = "not_a_dict"
    try:
        processor.cursor.execute(
            "UPDATE tasks SET metadata = %s WHERE task_id = %s;",
            (Json(invalid_data), "T004")
        )
    except Exception as e:
        # Expect an error due to invalid JSONB
        assert "invalid JSON" in str(e).lower()
        processor.conn.rollback()
    # Query metadata to ensure no corruption
    processor.cursor.execute(
        "SELECT metadata FROM tasks WHERE task_id = %s;",
        ("T004",)
    )
    result = processor.cursor.fetchone()
    # Verify original metadata is intact
    assert result[0] == {"priority": "medium", "owner": "ops_team"}

task_id = "T002"
metadata = {"priority": "low", "owner": "sales_team"}

query = "sales"

status = "Completed"

def query_metadata(cursor, owner):
    cursor.execute("SELECT metadata FROM tasks WHERE metadata.owner = %s;", (owner,))
    return cursor.fetchall()

from typing import Dict, Any
import psycopg2
from psycopg2.extras import Json

def update_metadata(task_id: str, metadata: Dict[str, Any]) -> None:
    """Update JSONB metadata for a task."""
    # Connect to tasks_db
    conn = psycopg2.connect(dbname="tasks_db", user="postgres", password="password", host="localhost")
    # Create a cursor
    cursor = conn.cursor()
    # Update metadata for the given task_id
    cursor.execute(
        "UPDATE tasks SET metadata = %s WHERE task_id = %s;",
        (Json(metadata), task_id)  # Json adapter converts dict to JSONB
    )
    # Commit the update
    conn.commit()
    # Log the update
    print(f"Updated metadata for {task_id}")
    # Close cursor and connection
    cursor.close()
    conn.close()

from typing import List, Tuple
import psycopg2

def search_with_ranking(query: str) -> List[Tuple[str, str, float]]:
    """Search tasks with ranking based on relevance."""
    # Connect to tasks_db
    conn = psycopg2.connect(dbname="tasks_db", user="postgres", password="password", host="localhost")
    # Create a cursor
    cursor = conn.cursor()
    # Search tasks and compute relevance rank
    cursor.execute(
        "SELECT task_id, description, ts_rank(search_vector, to_tsquery('english', %s)) AS rank "
        "FROM tasks WHERE search_vector @@ to_tsquery('english', %s) ORDER BY rank DESC;",
        (query, query)  # Query for the input term
    )
    # Fetch all results
    results = cursor.fetchall()
    # Print results for debugging
    print(results)
    # Close cursor and connection
    cursor.close()
    conn.close()
    # Return search results
    return results

import psycopg2

def create_filtered_view(status: str) -> None:
    """Create a materialized view for tasks with a specific status."""
    # Connect to tasks_db
    conn = psycopg2.connect(dbname="tasks_db", user="postgres", password="password", host="localhost")
    # Create a cursor
    cursor = conn.cursor()
    # Create materialized view for the given status
    cursor.execute("""
    CREATE MATERIALIZED VIEW IF NOT EXISTS filtered_summary AS
    SELECT status, COUNT(*) AS task_count
    FROM tasks
    WHERE status = %s
    GROUP BY status;
    """, (status,))
    # Commit the view creation
    conn.commit()
    # Log the creation
    print(f"Materialized view created for {status} tasks")
    # Close cursor and connection
    cursor.close()
    conn.close()

from typing import List, Dict, Any
import psycopg2

def query_metadata(cursor: psycopg2.cursor, owner: str) -> List[Dict[str, Any]]:
    """Query JSONB metadata by owner."""
    # Query tasks where owner matches the input
    cursor.execute(
        "SELECT metadata FROM tasks WHERE metadata->>'owner' = %s;",
        (owner,)  # Use ->> to get text value
    )
    # Return metadata, handling nulls with empty dict
    return [row[0] or {} for row in cursor.fetchall()]

from typing import List, Dict, Any
import psycopg2

def analyze_features() -> str:
    """Analyze performance of JSONB, full-text search, and materialized views."""
    # Connect to tasks_db
    conn = psycopg2.connect(dbname="tasks_db", user="postgres", password="password", host="localhost")
    # Create a cursor
    cursor = conn.cursor()
    # Query JSONB metadata
    cursor.execute("SELECT metadata->>'priority' FROM tasks WHERE metadata->>'owner' = %s;", ("finance_team",))
    # Query full-text search
    cursor.execute("SELECT task_id FROM tasks WHERE search_vector @@ to_tsquery('english', %s);", ("inventory",))
    # Query materialized view
    cursor.execute("SELECT task_count FROM task_status_summary WHERE status = %s;", ("Completed",))
    # Define performance explanation
    explanation = (
        "Explanation: JSONB enables flexible storage with O(log n) queries using GIN indexes. "
        "Full-text search provides O(log n) text queries with tsvector. Materialized views offer "
        "O(1) access for precomputed data but require O(n) refreshes. These features enable fast "
        "analytics for Hijra Group’s transaction tasks."
    )
    # Print explanation
    print(explanation)
    # Close cursor and connection
    cursor.close()
    conn.close()
    # Return explanation
    return explanation