基于PostgreSQL 10与Python 3的四表关联查询需求解决

Hey there! Let's tackle this query problem step by step. Since you're working with PostgreSQL 10 and Python 3, we can break this down into targeted SQL queries for each requirement, then wrap them in Python code to handle results cleanly.

First, let's align on table names (I'll use CompanyLogo as referenced in your problem statement, matching your example's Logo table structure).

1. Fetch all companies with up to 3 products and their first image #

We need to join all tables, limit each company to 3 products, and grab the first image (defined here as the smallest Id in ProductImage for each product) for each product. Here's the SQL using window functions to handle the limits and image selection:

WITH product_first_image AS (
    -- Get the first image (by Id) for each product
    SELECT 
        product_id,
        path AS product_image_path
    FROM (
        SELECT 
            product_id,
            path,
            ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY Id ASC) AS rn
        FROM ProductImage
    ) img
    WHERE rn = 1
),
company_products AS (
    -- Join all tables and assign row numbers to products per company
    SELECT 
        c.Id AS company_id,
        c.slug,
        cl.path AS logo_path,
        p.Id AS product_id,
        pfi.product_image_path,
        ROW_NUMBER() OVER (PARTITION BY c.Id ORDER BY p.Id ASC) AS product_rn
    FROM Company c
    JOIN CompanyLogo cl ON c.Id = cl.company_id
    LEFT JOIN Product p ON c.Id = p.company_id
    LEFT JOIN product_first_image pfi ON p.Id = pfi.product_id
)
-- Aggregate up to 3 products per company into a structured array
SELECT 
    company_id,
    slug,
    logo_path,
    ARRAY_AGG(
        JSON_BUILD_OBJECT(
            'product_id', product_id,
            'first_image', product_image_path
        ) ORDER BY product_id
    ) AS products
FROM company_products
WHERE product_rn <= 3 OR product_id IS NULL  -- Include companies with no products
GROUP BY company_id, slug, logo_path
ORDER BY company_id;

Quick breakdown: #

• product_first_image: CTE to isolate the first image for each product
• company_products: CTE to link companies to their logos and products, then tag each product with a row number per company (to cap at 3)
• Final query: Aggregates products into a JSON array for clean, structured results

2. Fetch a single company by ID with all products and their first image #

This is simpler since we don't need to limit product count. Here's the optimized SQL:

WITH product_first_image AS (
    SELECT 
        product_id,
        path AS product_image_path
    FROM (
        SELECT 
            product_id,
            path,
            ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY Id ASC) AS rn
        FROM ProductImage
    ) img
    WHERE rn = 1
)
SELECT 
    c.slug,
    cl.path AS logo_path,
    ARRAY_AGG(
        JSON_BUILD_OBJECT(
            'product_id', p.Id,
            'first_image', pfi.product_image_path
        ) ORDER BY p.Id
    ) AS products
FROM Company c
JOIN CompanyLogo cl ON c.Id = cl.company_id
LEFT JOIN Product p ON c.Id = p.company_id
LEFT JOIN product_first_image pfi ON p.Id = pfi.product_id
WHERE c.Id = %s  -- Parameter for company ID
GROUP BY c.Id, c.slug, cl.path;

Python Implementation (using psycopg2) #

Assuming you have psycopg2 installed (pip install psycopg2-binary), here's how to execute these queries and process results into Python-friendly dictionaries:

Fetch all companies with up to 3 products #

import psycopg2
from psycopg2.extras import RealDictCursor

def get_all_companies():
    # Connect to your PostgreSQL database
    conn = psycopg2.connect(
        dbname="your_db_name",
        user="your_db_user",
        password="your_db_password",
        host="your_db_host"
    )
    # Use RealDictCursor to get results as dictionaries
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    query = """
    WITH product_first_image AS (
        SELECT 
            product_id,
            path AS product_image_path
        FROM (
            SELECT 
                product_id,
                path,
                ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY Id ASC) AS rn
            FROM ProductImage
        ) img
        WHERE rn = 1
    ),
    company_products AS (
        SELECT 
            c.Id AS company_id,
            c.slug,
            cl.path AS logo_path,
            p.Id AS product_id,
            pfi.product_image_path,
            ROW_NUMBER() OVER (PARTITION BY c.Id ORDER BY p.Id ASC) AS product_rn
        FROM Company c
        JOIN CompanyLogo cl ON c.Id = cl.company_id
        LEFT JOIN Product p ON c.Id = p.company_id
        LEFT JOIN product_first_image pfi ON p.Id = pfi.product_id
    )
    SELECT 
        company_id,
        slug,
        logo_path,
        ARRAY_AGG(
            JSON_BUILD_OBJECT(
                'product_id', product_id,
                'first_image', product_image_path
            ) ORDER BY product_id
        ) AS products
    FROM company_products
    WHERE product_rn <= 3 OR product_id IS NULL
    GROUP BY company_id, slug, logo_path
    ORDER BY company_id;
    """
    
    cursor.execute(query)
    companies = cursor.fetchall()
    
    # Clean up connections
    cursor.close()
    conn.close()
    
    # Convert PostgreSQL JSON objects to Python dictionaries
    for company in companies:
        company['products'] = [dict(item) for item in company['products'] if item is not None]
    
    return companies

Fetch single company by ID #

def get_company_by_id(company_id):
    conn = psycopg2.connect(
        dbname="your_db_name",
        user="your_db_user",
        password="your_db_password",
        host="your_db_host"
    )
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    
    query = """
    WITH product_first_image AS (
        SELECT 
            product_id,
            path AS product_image_path
        FROM (
            SELECT 
                product_id,
                path,
                ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY Id ASC) AS rn
            FROM ProductImage
        ) img
        WHERE rn = 1
    )
    SELECT 
        c.slug,
        cl.path AS logo_path,
        ARRAY_AGG(
            JSON_BUILD_OBJECT(
                'product_id', p.Id,
                'first_image', pfi.product_image_path
            ) ORDER BY p.Id
        ) AS products
    FROM Company c
    JOIN CompanyLogo cl ON c.Id = cl.company_id
    LEFT JOIN Product p ON c.Id = p.company_id
    LEFT JOIN product_first_image pfi ON p.Id = pfi.product_id
    WHERE c.Id = %s
    GROUP BY c.Id, c.slug, cl.path;
    """
    
    # Pass the company ID as a parameter to avoid SQL injection
    cursor.execute(query, (company_id,))
    company = cursor.fetchone()
    
    cursor.close()
    conn.close()
    
    if company:
        company['products'] = [dict(item) for item in company['products'] if item is not None]
    
    return company

Key notes: #

• RealDictCursor returns results as Python dictionaries, making them easy to work with in your application
• Using ARRAY_AGG and JSON_BUILD_OBJECT in PostgreSQL keeps most of the data structuring work in the database, reducing Python-side processing
• The queries handle companies with no products gracefully, returning an empty products list for them

内容的提问来源于stack exchange，提问作者user3541631