Tối Ưu Hóa Truy Vấn SQL Nâng Cao Trong Python: Từ CTEs Đến Window Functions

import psycopg2
from psycopg2.extras import RealDictCursor

# Định nghĩa cấu hình kết nối cơ sở dữ liệu
DATABASE_CONFIG = {
    "dbname": "ecommerce_db",
    "user": "postgres",
    "password": "secure_password",
    "host": "localhost",
    "port": 5432
}

def fetch_high_performing_products(min_revenue_threshold):
    # Câu lệnh SQL sử dụng nhiều CTE phức tạp
    advanced_query = """
    WITH CategoryRevenue AS (
        -- CTE thứ nhất: Tính tổng doanh thu của từng sản phẩm
        SELECT 
            p.product_id,
            p.product_name,
            p.category_id,
            SUM(s.quantity * s.unit_price) AS total_revenue
        FROM products p
        JOIN sales s ON p.product_id = s.product_id
        GROUP BY p.product_id, p.product_name, p.category_id
    ),
    AverageCategorySales AS (
        -- CTE thứ hai: Tính doanh thu trung bình của từng danh mục
        SELECT 
            category_id,
            AVG(total_revenue) AS avg_category_revenue
        FROM CategoryRevenue
        GROUP BY category_id
    )
    -- Truy vấn chính kết hợp các CTE và tham số đầu vào từ Python
    SELECT 
        cr.product_name,
        cr.total_revenue,
        acs.avg_category_revenue
    FROM CategoryRevenue cr
    JOIN AverageCategorySales acs ON cr.category_id = acs.category_id
    WHERE cr.total_revenue > acs.avg_category_revenue
      AND cr.total_revenue >= %s
    ORDER BY cr.total_revenue DESC;
    """

    connection = None
    try:
        # Thiết lập kết nối đến cơ sở dữ liệu
        connection = psycopg2.connect(**DATABASE_CONFIG)
        with connection.cursor(cursor_factory=RealDictCursor) as cursor:
            # Thực thi truy vấn với tham số an toàn chống SQL Injection
            cursor.execute(advanced_query, (min_revenue_threshold,))
            query_results = cursor.fetchall()
            return query_results
    except Exception as database_error:
        print(f"Đã xảy ra lỗi khi thực thi truy vấn: {database_error}")
        return []
    finally:
        if connection:
            connection.close()

# Thực thi hàm với ngưỡng doanh thu tối thiểu
if __name__ == "__main__":
    threshold_value = 50000.00
    results = fetch_high_performing_products(threshold_value)
    for item in results:
        print(f"Sản phẩm: {item['product_name']} - Doanh thu: {item['total_revenue']}")

import sqlite3

def get_top_employees_by_department(database_path):
    connection = sqlite3.connect(database_path)
    cursor = connection.cursor()

    # Câu lệnh SQL sử dụng Window Function RANK()
    ranking_query = """
    WITH RankedEmployees AS (
        SELECT 
            employee_name,
            department_name,
            sales_amount,
            RANK() OVER (
                PARTITION BY department_name 
                ORDER BY sales_amount DESC
            ) as sales_rank
        FROM employee_sales
    )
    SELECT 
        employee_name,
        department_name,
        sales_amount
    FROM RankedEmployees
    WHERE sales_rank <= 3;
    """

    cursor.execute(ranking_query)
    department_rankings = cursor.fetchall()
    
    connection.close()
    return department_rankings

import psycopg2
from psycopg2.extras import RealDictCursor

DATABASE_CONFIG = {
    "dbname": "retail_db",
    "user": "postgres",
    "password": "secure_password",
    "host": "localhost",
    "port": 5432
}

def get_running_revenue_by_store(target_store_id):
    # Câu lệnh SQL sử dụng Window Function với khung cửa sổ (Frame Clause)
    running_total_query = """
    SELECT 
        sale_date,
        daily_revenue,
        SUM(daily_revenue) OVER (
            PARTITION BY store_id 
            ORDER BY sale_date 
            ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
        ) AS cumulative_revenue
    FROM daily_store_sales
    WHERE store_id = %s
    ORDER BY sale_date ASC;
    """

    connection = None
    try:
        connection = psycopg2.connect(**DATABASE_CONFIG)
        with connection.cursor(cursor_factory=RealDictCursor) as cursor:
            # Truyền tham số store_id an toàn vào câu lệnh
            cursor.execute(running_total_query, (target_store_id,))
            execution_results = cursor.fetchall()
            return execution_results
    except Exception as query_error:
        print(f"Lỗi khi thực thi truy vấn tổng tích lũy: {query_error}")
        return []
    finally:
        if connection:
            connection.close()

# Thực thi hàm thử nghiệm cho cửa hàng có ID bằng 101
if __name__ == "__main__":
    store_id_param = 101
    results = get_running_revenue_by_store(store_id_param)
    for record in results:
        print(f"Ngày: {record['sale_date']} | Doanh thu ngày: {record['daily_revenue']} | Tích lũy: {record['cumulative_revenue']}")

import psycopg2

def fetch_subcategories_hierarchy(root_category_id):
    # Câu lệnh SQL sử dụng Recursive CTE để duyệt cây phân cấp
    recursive_query = """
    WITH RECURSIVE CategoryTree AS (
        -- Bước neo (Anchor Member): Lấy danh mục gốc ban đầu
        SELECT 
            category_id, 
            category_name, 
            parent_id, 
            1 AS hierarchy_level
        FROM categories
        WHERE category_id = %s
        
        UNION ALL
        
        -- Bước đệ quy (Recursive Member): Lấy các danh mục con liên kết cấp tiếp theo
        SELECT 
            c.category_id, 
            c.category_name, 
            c.parent_id, 
            ct.hierarchy_level + 1
        FROM categories c
        JOIN CategoryTree ct ON c.parent_id = ct.category_id
    )
    -- Truy vấn tổng hợp từ kết quả của Recursive CTE
    SELECT 
        category_id, 
        category_name, 
        parent_id, 
        hierarchy_level
    FROM CategoryTree
    ORDER BY hierarchy_level ASC, category_name ASC;
    """

    database_connection = psycopg2.connect(
        dbname="ecommerce_db",
        user="postgres",
        password="secure_password",
        host="localhost",
        port=5432
    )
    
    try:
        with database_connection.cursor() as cursor:
            cursor.execute(recursive_query, (root_category_id,))
            category_hierarchy_data = cursor.fetchall()
            return category_hierarchy_data
    except Exception as error_detail:
        print(f"Đã xảy ra lỗi hệ thống khi quét phân cấp danh mục: {error_detail}")
        return []
    finally:
        database_connection.close()