如何在SQLAlchemy中自连接表并计算收入与增长率？

Got it, let's break down how to achieve your desired result using SQLAlchemy. Your goal is to pivot the annual income data for Company A into year-based columns, plus add a row for the year-over-year growth ratio. Here's a step-by-step solution:

1. Define Your ORM Model #

First, let's set up the model that maps to your company_income table:

from sqlalchemy import Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class CompanyIncome(Base):
    __tablename__ = 'company_income'
    id = Column(Integer, primary_key=True)
    company_name = Column(String)
    income = Column(Float)
    year = Column(Integer)

2. Calculate Income and Growth Ratios #

We'll use SQL's LAG window function to fetch the previous year's income, then compute the growth ratio. We wrap this in a subquery to reuse the data later:

from sqlalchemy import select, func, case, literal_column, union_all
from sqlalchemy.orm import sessionmaker
import sqlalchemy as sa

# Assume you've already created your database engine
Session = sessionmaker(bind=engine)
session = Session()

# Subquery to get each year's income and the prior year's income
income_subquery = select(
    CompanyIncome.year,
    CompanyIncome.income,
    # Grab the previous year's income using the LAG window function
    func.lag(CompanyIncome.income).over(order_by=CompanyIncome.year).label('prev_income')
).where(CompanyIncome.company_name == 'A').subquery()

# Calculate growth ratio (handle 2010's missing prior year with '-')
ratio_subquery = select(
    income_subquery.c.year,
    income_subquery.c.income,
    case(
        (income_subquery.c.prev_income.is_(None), literal_column("'-'")),
        else_=(income_subquery.c.income / income_subquery.c.prev_income).cast(Float)
    ).label('increase_ratio')
).subquery()

3. Pivot Rows to Columns (Static Year Example) #

If you know the exact years you need (2010-2012), we use conditional aggregation to pivot the data into columns, then combine the two rows (income and growth ratio):

# Build the "income" row with year columns
income_row = select(
    literal_column("'income'").label('metric'),
    func.sum(case((ratio_subquery.c.year == 2010, ratio_subquery.c.income), else_=None)).label('2010'),
    func.sum(case((ratio_subquery.c.year == 2011, ratio_subquery.c.income), else_=None)).label('2011'),
    func.sum(case((ratio_subquery.c.year == 2012, ratio_subquery.c.income), else_=None)).label('2012')
)

# Build the "increase_ratio" row with year columns
ratio_row = select(
    literal_column("'increase_ratio'").label('metric'),
    func.sum(case((ratio_subquery.c.year == 2010, ratio_subquery.c.increase_ratio), else_=None)).label('2010'),
    func.sum(case((ratio_subquery.c.year == 2011, ratio_subquery.c.increase_ratio), else_=None)).label('2011'),
    func.sum(case((ratio_subquery.c.year == 2012, ratio_subquery.c.increase_ratio), else_=None)).label('2012')
)

# Combine the two rows and order them to match your desired output
final_query = union_all(income_row, ratio_row).order_by(literal_column("metric"))

# Execute the query and fetch results
results = session.execute(final_query).fetchall()

# Optional: Convert to a DataFrame for cleaner display
import pandas as pd
df = pd.DataFrame(results, columns=['', '2010', '2011', '2012'])
print(df)

4. Dynamic Pivot (For Variable Years) #

If you don't want to hardcode years, you can dynamically fetch all years for Company A and build the pivot columns:

# Get all unique years for Company A
years = session.query(CompanyIncome.year).filter(CompanyIncome.company_name == 'A').order_by(CompanyIncome.year).all()
year_list = [year[0] for year in years]

# Dynamically create income columns for each year
income_columns = [
    func.sum(case((ratio_subquery.c.year == y, ratio_subquery.c.income), else_=None)).label(str(y))
    for y in year_list
]

# Dynamically create ratio columns for each year
ratio_columns = [
    func.sum(case((ratio_subquery.c.year == y, ratio_subquery.c.increase_ratio), else_=None)).label(str(y))
    for y in year_list
]

# Build rows with dynamic columns
income_row = select(literal_column("'income'").label('metric'), *income_columns)
ratio_row = select(literal_column("'increase_ratio'").label('metric'), *ratio_columns)

final_query = union_all(income_row, ratio_row).order_by(literal_column("metric"))

Key Notes: #

• Window Function (LAG): Eliminates the need for explicit joins to get the previous year's income, making the query cleaner.
• Conditional Aggregation: Uses SUM + CASE to pivot rows into columns—this is SQLAlchemy's way of creating a pivot table.
• Union: Combines the two separate result sets (income and growth ratio) into a single table matching your desired output.

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