pandas for Backend Developers: Logs, Excel/ERP, ETL, and Reports — Why I Didn't Use It Earlier

# Task: find top-10 endpoints by response time for the last week
# Option 1: Pure Python (15+ lines with loops, dict comprehensions)
# Option 2: SQL query + ORM (need to write migration for analytics table)
# Option 3: pandas (3 lines of code)
 
import pandas as pd
 
df = pd.read_csv('api_logs.csv')
top_slow = df.nlargest(10, 'response_time')[['endpoint', 'response_time']]
print(top_slow)

# ❌ BAD: writing to DB via pandas
df.to_sql('users', engine, if_exists='append')
# Problems: slow, no transaction support, complex error handling
 
# ✅ GOOD: use ORM or bulk insert
session.bulk_insert_mappings(User, records)

# ❌ BAD: loading 10GB file
df = pd.read_csv('huge_file.csv')  # OOM (Out of Memory)
 
# ✅ GOOD: use chunked reading, Dask, or DuckDB
for chunk in pd.read_csv('huge_file.csv', chunksize=10000):
    process_chunk(chunk)

# ❌ BAD: pandas in FastAPI endpoint for each request
@app.get("/stats")
def get_stats():
    df = pd.read_sql("SELECT * FROM orders", engine)  # Slow!
    return df.groupby('product').sum().to_dict()
 
# ✅ GOOD: pre-calculate aggregates or use SQL/Redis

# ❌ BAD: modifying data via pandas with DB save
df = pd.read_sql("SELECT * FROM accounts", engine)
df.loc[df['id'] == 123, 'balance'] -= 100  # No transaction!
df.to_sql('accounts', engine, if_exists='replace')  # Data loss on error, no audit
 
# ✅ GOOD: use SQL transactions with operation logging
with session.begin():
    # Lock account for update
    account = session.query(Account).filter_by(id=123).with_for_update().first()
    account.balance -= 100
 
    # Record transaction for audit
    transaction = Transaction(
        account_id=123,
        amount=-100,
        type='withdrawal',
        timestamp=datetime.now()
    )
    session.add(transaction)

# ❌ BAD: loading everything into pandas for simple filtering
df = pd.read_sql("SELECT * FROM users", engine)
active_users = df[df['is_active'] == True]
 
# ✅ GOOD: use SQL
df = pd.read_sql("SELECT * FROM users WHERE is_active = TRUE", engine)

import pandas as pd
 
# Creating DataFrame from dict
data = {
    'user_id': [1, 2, 3, 4],
    'name': ['Alice', 'Bob', 'Charlie', 'Diana'],
    'age': [25, 30, 35, 28],
    'city': ['Moscow', 'SPb', 'Moscow', 'Kazan']
}
 
df = pd.DataFrame(data)
print(df)
 
#    user_id     name  age     city
# 0        1    Alice   25   Moscow
# 1        2      Bob   30      SPb
# 2        3  Charlie   35   Moscow
# 3        4    Diana   28    Kazan

# Getting a column
ages = df['age']
print(type(ages))  # <class 'pandas.core.series.Series'>
 
# Operations on Series
print(ages.mean())     # 29.5 (average)
print(ages.max())      # 35
print(ages.std())      # 4.2 (standard deviation)

# Filtering
moscow_users = df[df['city'] == 'Moscow']
 
# Sorting
sorted_by_age = df.sort_values('age', ascending=False)
 
# Selecting columns
names_and_ages = df[['name', 'age']]
 
# Adding column
df['is_adult'] = df['age'] >= 18
 
# Grouping and aggregation
city_stats = df.groupby('city')['age'].agg(['mean', 'count'])

import pandas as pd
from datetime import datetime
 
# 1. Reading logs (assume they're in CSV)
df = pd.read_csv(
    'access.log',
    sep='|',
    names=['timestamp', 'method_path', 'status', 'response_time'],
    parse_dates=['timestamp']  # Automatic date parsing
)
 
# 2. Data cleaning (remove whitespace!)
df['response_time'] = df['response_time'].str.strip().str.replace('ms', '').astype(int)
df[['method', 'endpoint']] = df['method_path'].str.strip().str.split(' ', expand=True)
df['status'] = df['status'].astype(int)
# .str.strip() is critical — logs often contain spaces around separators
 
# 3. Top-10 slowest endpoints
top_slow = (
    df.groupby('endpoint')['response_time']
    .agg(['mean', 'max', 'count'])
    .sort_values('mean', ascending=False)
    .head(10)
)
 
print("Top-10 slowest endpoints:")
print(top_slow)
 
# 4. Endpoints with high percentage of 5xx errors (vectorized!)
error_rate = (
    df.assign(is_5xx=df['status'] >= 500)
    .groupby('endpoint')['is_5xx']
    .mean()
    .mul(100)
    .sort_values(ascending=False)
    .head(10)
)
 
print("\nEndpoints with high % of errors:")
print(error_rate)
 
# 5. Response time distribution by hours
df['hour'] = df['timestamp'].dt.hour
hourly_stats = df.groupby('hour')['response_time'].agg(['mean', 'median', 'count'])
 
print("\nDistribution by hours:")
print(hourly_stats)

import matplotlib.pyplot as plt
 
# Plot average response time by hours
hourly_stats['mean'].plot(kind='line', title='Average Response Time by Hour')
plt.xlabel('Hour of Day')
plt.ylabel('Average Time (ms)')
plt.savefig('response_time_by_hour.png')

import pandas as pd
 
# 1. Loading data from Excel (different sheets)
sales = pd.read_excel('sales.xlsx', sheet_name='Sales')
stock = pd.read_excel('sales.xlsx', sheet_name='Inventory')
costs = pd.read_excel('sales.xlsx', sheet_name='Costs')
 
# 2. Cleaning column names (ERPs love spaces and special characters)
sales.columns = sales.columns.str.strip()
stock.columns = stock.columns.str.strip()
costs.columns = costs.columns.str.strip()
 
# 3. Merging data
# Adding costs to sales
df = sales.merge(
    costs[['SKU', 'Cost']],
    on='SKU',
    how='left'
)
 
# 4. Margin calculation (with division by zero protection)
df['Margin'] = df['Sale Price'] - df['Cost']
df['Margin %'] = (
    df['Margin'] / df['Sale Price'].replace(0, float('nan')) * 100
).round(2)
 
# 5. Analysis: top products by margin
top_margin = (
    df.groupby(['Category', 'Product Name'])
    .agg({
        'Quantity': 'sum',
        'Margin': 'sum',
        'Margin %': 'mean'
    })
    .sort_values('Margin', ascending=False)
    .head(20)
)
 
print("Top-20 products by margin:")
print(top_margin)
 
# 6. Analysis: unprofitable products
unprofitable = df[df['Margin'] < 0].groupby('Category').agg({
    'Product Name': 'count',
    'Margin': 'sum'
})
 
print("\nUnprofitable products by categories:")
print(unprofitable)
 
# 7. Pivot table for management (requires "Date" column in source data!)
pivot = df.pivot_table(
    values=['Quantity', 'Margin'],
    index='Category',
    columns=pd.to_datetime(df['Date']).dt.month,
    aggfunc='sum',
    fill_value=0
)
 
print("\nSales and margin by months:")
print(pivot)
 
# 8. Export results to new Excel
with pd.ExcelWriter('analysis_results.xlsx') as writer:
    top_margin.to_excel(writer, sheet_name='Top Products')
    unprofitable.to_excel(writer, sheet_name='Unprofitable')
    pivot.to_excel(writer, sheet_name='Monthly Dynamics')
 
print("\n✅ Analysis complete: analysis_results.xlsx")

import logging
from pathlib import Path
from datetime import datetime
from typing import Optional
import pandas as pd
 
# Logging setup
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
 
class SalesAnalysisError(Exception):
    """Error during sales analysis."""
    pass
 
def analyze_erp_sales(
    file_path: str,
    output_dir: str = 'reports',
    required_sheets: Optional[list] = None
) -> str:
    """
    Automatic sales analysis from ERP export with error handling.
 
    Args:
        file_path: Path to Excel file with ERP export
        output_dir: Directory for saving report
        required_sheets: List of required sheets (default: Sales, Costs)
 
    Returns:
        Path to created report
 
    Raises:
        SalesAnalysisError: On validation or processing errors
    """
    if required_sheets is None:
        required_sheets = ['Sales', 'Costs']
 
    try:
        logger.info(f"Starting analysis of file: {file_path}")
 
        # 1. File existence check
        if not Path(file_path).exists():
            raise SalesAnalysisError(f"File not found: {file_path}")
 
        # 2. Create reports directory
        Path(output_dir).mkdir(parents=True, exist_ok=True)
 
        # 3. Load data with validation
        try:
            excel_file = pd.ExcelFile(file_path)
            available_sheets = excel_file.sheet_names
 
            # Check required sheets presence
            missing_sheets = set(required_sheets) - set(available_sheets)
            if missing_sheets:
                raise SalesAnalysisError(
                    f"Missing required sheets: {missing_sheets}. "
                    f"Available sheets: {available_sheets}"
                )
 
            sales = pd.read_excel(excel_file, sheet_name='Sales')
            costs = pd.read_excel(excel_file, sheet_name='Costs')
 
        except ValueError as e:
            raise SalesAnalysisError(f"Error reading Excel file: {e}")
 
        # 4. Data structure validation
        required_sales_cols = ['SKU', 'Quantity', 'Sale Price', 'Category', 'Product Name']
        required_costs_cols = ['SKU', 'Cost']
 
        missing_sales_cols = set(required_sales_cols) - set(sales.columns)
        missing_costs_cols = set(required_costs_cols) - set(costs.columns)
 
        if missing_sales_cols:
            raise SalesAnalysisError(f"'Sales' sheet missing columns: {missing_sales_cols}")
        if missing_costs_cols:
            raise SalesAnalysisError(f"'Costs' sheet missing columns: {missing_costs_cols}")
 
        logger.info(f"Loaded {len(sales)} sales records, {len(costs)} cost records")
 
        # 5. Data cleaning
        sales.columns = sales.columns.str.strip()
        costs.columns = costs.columns.str.strip()
 
        # Remove empty rows
        sales = sales.dropna(subset=['SKU'])
        costs = costs.dropna(subset=['SKU'])
 
        # 6. Merging and margin calculation
        df = sales.merge(
            costs[['SKU', 'Cost']],
            on='SKU',
            how='left'
        )
 
        # Check for products without costs
        missing_costs = df['Cost'].isna().sum()
        if missing_costs > 0:
            logger.warning(f"Found {missing_costs} products without costs")
            df['Cost'] = df['Cost'].fillna(0)
 
        df['Margin'] = df['Sale Price'] - df['Cost']
        # Division by zero protection
        df['Margin %'] = (
            df['Margin'] / df['Sale Price'].replace(0, float('nan')) * 100
        ).round(2)
 
        # 7. Analysis
        top_margin = df.groupby(['Category', 'Product Name']).agg({
            'Quantity': 'sum',
            'Margin': 'sum',
            'Margin %': 'mean'
        }).sort_values('Margin', ascending=False).head(20)
 
        unprofitable = df[df['Margin'] < 0].groupby('Category').agg({
            'Product Name': 'count',
            'Margin': 'sum'
        })
 
        # 8. Report export
        timestamp = datetime.now().strftime('%Y%m%d_%H%M')
        output_file = Path(output_dir) / f"sales_analysis_{timestamp}.xlsx"
 
        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
            top_margin.to_excel(writer, sheet_name='Top Products')
 
            if not unprofitable.empty:
                unprofitable.to_excel(writer, sheet_name='Unprofitable')
            else:
                logger.info("No unprofitable products found")
 
            # Summary statistics
            summary = pd.DataFrame({
                'Metric': [
                    'Total Sales',
                    'Total Margin',
                    'Average Margin %',
                    'Unprofitable Products'
                ],
                'Value': [
                    f"{df['Quantity'].sum():,.0f} units",
                    f"${df['Margin'].sum():,.2f}",
                    f"{df['Margin %'].mean():.2f}%",
                    len(df[df['Margin'] < 0])
                ]
            })
            summary.to_excel(writer, sheet_name='Summary', index=False)
 
        logger.info(f"✅ Report successfully created: {output_file}")
        return str(output_file)
 
    except SalesAnalysisError:
        raise
    except Exception as e:
        logger.error(f"Unexpected error during analysis: {e}", exc_info=True)
        raise SalesAnalysisError(f"Data processing error: {e}")
 
# Usage with error handling
try:
    report = analyze_erp_sales('sales.xlsx')
    print(f"✅ Report ready: {report}")
except SalesAnalysisError as e:
    print(f"❌ Error: {e}")
    # Send notification to admin
except Exception as e:
    print(f"❌ Critical error: {e}")
    # Log to Sentry/other monitoring system

import pandas as pd
from sqlalchemy import create_engine
 
# 1. Database connection
engine = create_engine('postgresql://user:password@localhost:5432/mydb')
 
# 2. Data loading
query = """
    SELECT
        u.id,
        u.email,
        u.created_at,
        u.city,
        COUNT(o.id) as orders_count,
        SUM(o.total) as total_spent
    FROM users u
    LEFT JOIN orders o ON u.id = o.user_id
    WHERE u.created_at >= '2025-01-01'
    GROUP BY u.id, u.email, u.created_at, u.city
"""
 
df = pd.read_sql(query, engine)
 
# 3. Data transformation
df['created_at'] = pd.to_datetime(df['created_at'])
df['registration_month'] = df['created_at'].dt.to_period('M')
 
# User categorization (vectorized approach)
import numpy as np
 
# ✅ Using np.select instead of apply() — 10-100x faster!
conditions = [
    df['total_spent'] > 10000,      # Condition for VIP
    df['orders_count'] > 5,          # Condition for Active
    df['orders_count'] > 0           # Condition for Buyer
]
choices = ['VIP', 'Active', 'Buyer']
df['category'] = np.select(conditions, choices, default='Registered')
 
# Why is this faster?
# apply() calls Python function for EACH row (row by row)
# np.select() performs operation over entire array at once in C/NumPy
# On 100k rows: apply() ~2 sec, np.select() ~20 ms
 
# 4. Adding data from external source (e.g., city -> region)
city_to_region = {
    'Moscow': 'Central',
    'SPb': 'Northwest',
    'Kazan': 'Volga'
}
df['region'] = df['city'].map(city_to_region)
 
# 5. Export to Excel with formatting
with pd.ExcelWriter('users_report.xlsx', engine='openpyxl') as writer:
    # Main report
    df.to_excel(writer, sheet_name='Users', index=False)
 
    # Summary table by categories
    summary = df.groupby('category').agg({
        'id': 'count',
        'total_spent': 'sum'
    }).rename(columns={'id': 'users_count'})
 
    summary.to_excel(writer, sheet_name='Summary')

import pandas as pd
from pathlib import Path
from datetime import datetime, timedelta
 
def daily_etl_pipeline(date: datetime):
    """
    Daily ETL: loads data for a day, processes, saves.
    """
    # 1. Data loading
    raw_data_path = f'raw_data/{date.strftime("%Y-%m-%d")}.csv'
    df = pd.read_csv(raw_data_path)
 
    # 2. Validation
    required_columns = ['user_id', 'event_type', 'timestamp']
    missing = set(required_columns) - set(df.columns)
    if missing:
        raise ValueError(f"Missing columns: {missing}")
 
    # 3. Cleaning
    df = df.dropna(subset=['user_id'])  # Remove rows without user_id
    df = df.drop_duplicates()           # Remove duplicates
 
    # 4. Type conversion
    df['timestamp'] = pd.to_datetime(df['timestamp'])
    df['user_id'] = df['user_id'].astype(int)
 
    # 5. Filter anomalies
    df = df[df['timestamp'].dt.date == date.date()]  # Only data for needed day
 
    # 6. Aggregation
    daily_stats = df.groupby('user_id').agg({
        'event_type': 'count',
        'timestamp': ['min', 'max']
    }).reset_index()
 
    # 7. Save result
    output_path = f'processed_data/{date.strftime("%Y-%m-%d")}_stats.parquet'
    daily_stats.to_parquet(output_path, compression='snappy')
 
    return daily_stats
 
# Run pipeline
yesterday = datetime.now() - timedelta(days=1)
stats = daily_etl_pipeline(yesterday)
print(f"Processed {len(stats)} users")

import pandas as pd
import re
 
def validate_and_clean_csv(file_path: str) -> pd.DataFrame:
    """
    Validation and cleaning of CSV with clients.
    """
 
    # 1. Loading
    df = pd.read_csv(file_path)
 
    # 2. Check required columns
    required = ['name', 'email', 'phone']
    missing = set(required) - set(df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")
 
    # 3. Remove empty rows
    df = df.dropna(subset=required)
 
    # 4. Email validation (vectorized!)
    email_pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    df['email_valid'] = df['email'].str.match(email_pattern, na=False)
 
    invalid_emails = df[~df['email_valid']]
    if len(invalid_emails) > 0:
        print(f"⚠️ Found {len(invalid_emails)} invalid emails:")
        print(invalid_emails[['name', 'email']].head())
 
    df = df[df['email_valid']].drop('email_valid', axis=1)
 
    # 5. Phone cleaning (bring to unified format)
    def clean_phone(phone):
        # Remove everything except digits
        digits = re.sub(r'\D', '', str(phone))
        # Add country code if needed
        if digits.startswith('8') and len(digits) == 11:
            digits = '1' + digits[1:]
        elif digits.startswith('9') and len(digits) == 10:
            digits = '1' + digits
        return f"+{digits}"
 
    df['phone'] = df['phone'].apply(clean_phone)
 
    # 6. Remove duplicates by email
    duplicates = df[df.duplicated(subset=['email'], keep=False)]
    if len(duplicates) > 0:
        print(f"⚠️ Found {len(duplicates)} duplicate emails")
        print(duplicates[['name', 'email']])
 
    df = df.drop_duplicates(subset=['email'], keep='first')
 
    # 7. Bring name to title case
    df['name'] = df['name'].str.strip().str.title()
 
    return df
 
# Usage
try:
    clean_df = validate_and_clean_csv('clients.csv')
    clean_df.to_csv('clients_clean.csv', index=False)
    print(f"✅ Successfully cleaned {len(clean_df)} records")
except Exception as e:
    print(f"❌ Validation failed: {e}")

def create_validation_report(df: pd.DataFrame) -> dict:
    """
    Creates detailed data quality report.
    """
 
    report = {
        'total_rows': len(df),
        'missing_values': df.isnull().sum().to_dict(),
        'duplicates': {
            'email': df['email'].duplicated().sum(),
            'phone': df['phone'].duplicated().sum()
        },
        'data_types': df.dtypes.astype(str).to_dict(),
        'unique_values': {
            col: df[col].nunique()
            for col in df.columns
        }
    }
 
    # Statistics for numeric columns
    numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns
    if len(numeric_cols) > 0:
        report['numeric_stats'] = df[numeric_cols].describe().to_dict()
 
    # Top-10 most frequent values for categorical columns
    categorical_cols = df.select_dtypes(include=['object']).columns
    report['top_values'] = {
        col: df[col].value_counts().head(10).to_dict()
        for col in categorical_cols
    }
 
    return report
 
# Generate report
report = create_validation_report(df)
print(f"Total rows: {report['total_rows']}")
print(f"Missing values: {report['missing_values']}")
print(f"Email duplicates: {report['duplicates']['email']}")

import pandas as pd
from sqlalchemy import create_engine
from datetime import datetime
 
def generate_monthly_report(year: int, month: int):
    """
    Generates monthly business report.
    """
 
    engine = create_engine('postgresql://user:password@localhost:5432/mydb')
 
    # 1. Load data for month (safe parameterized query)
    from datetime import datetime
    from dateutil.relativedelta import relativedelta
 
    # Create period boundaries
    start_date = datetime(year, month, 1)
    end_date = start_date + relativedelta(months=1)
 
    # ✅ Parameterized query (SQL injection protection)
    query = """
        SELECT
            o.id as order_id,
            o.created_at,
            o.total,
            o.status,
            u.id as user_id,
            u.email,
            u.city
        FROM orders o
        JOIN users u ON o.user_id = u.id
        WHERE o.created_at >= :start_date AND o.created_at < :end_date
    """
 
    df = pd.read_sql(query, engine, params={'start_date': start_date, 'end_date': end_date})
    df['created_at'] = pd.to_datetime(df['created_at'])
    df['date'] = df['created_at'].dt.date
 
    # 2. Create Excel with multiple sheets
    output_file = f'monthly_report_{year}_{month:02d}.xlsx'
 
    with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
        # Sheet 1: Overall statistics
        summary = pd.DataFrame({
            'Metric': [
                'Total Orders',
                'Sales Sum',
                'Average Check',
                'Unique Customers',
                'Completed Orders',
                'Cancelled Orders'
            ],
            'Value': [
                len(df),
                f"${df['total'].sum():,.2f}",
                f"${df['total'].mean():,.2f}",
                df['user_id'].nunique(),
                len(df[df['status'] == 'completed']),
                len(df[df['status'] == 'cancelled'])
            ]
        })
        summary.to_excel(writer, sheet_name='Overall Statistics', index=False)
 
        # Sheet 2: Top-10 clients
        top_clients = (
            df.groupby(['user_id', 'email'])
            .agg({
                'order_id': 'count',
                'total': 'sum'
            })
            .rename(columns={
                'order_id': 'Order Count',
                'total': 'Total Spent'
            })
            .sort_values('Total Spent', ascending=False)
            .head(10)
            .reset_index()
        )
        top_clients.to_excel(writer, sheet_name='Top-10 Clients', index=False)
 
        # Sheet 3: Daily dynamics
        daily_sales = (
            df.groupby('date')
            .agg({
                'order_id': 'count',
                'total': 'sum'
            })
            .rename(columns={
                'order_id': 'Orders',
                'total': 'Revenue'
            })
            .reset_index()
        )
        daily_sales.to_excel(writer, sheet_name='Daily Dynamics', index=False)
 
        # Sheet 4: Breakdown by cities
        city_stats = (
            df.groupby('city')
            .agg({
                'order_id': 'count',
                'total': ['sum', 'mean'],
                'user_id': 'nunique'
            })
            .round(2)
        )
        city_stats.columns = ['Orders', 'Revenue', 'Avg Check', 'Customers']
        city_stats.to_excel(writer, sheet_name='By Cities')
 
    print(f"✅ Report saved: {output_file}")
    return output_file
 
# Generate report
report_file = generate_monthly_report(2025, 12)

from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Alignment
 
def format_excel_report(file_path: str):
    """
    Adds formatting to Excel report.
    """
 
    wb = load_workbook(file_path)
 
    # Formatting for all sheets
    for sheet_name in wb.sheetnames:
        ws = wb[sheet_name]
 
        # Headers: bold font, gray background
        header_fill = PatternFill(start_color="CCCCCC", end_color="CCCCCC", fill_type="solid")
        header_font = Font(bold=True)
 
        for cell in ws[1]:  # First row (headers)
            cell.fill = header_fill
            cell.font = header_font
            cell.alignment = Alignment(horizontal='center')
 
        # Auto-width columns
        for column in ws.columns:
            max_length = 0
            column_letter = column[0].column_letter
            for cell in column:
                try:
                    if len(str(cell.value)) > max_length:
                        max_length = len(str(cell.value))
                except:
                    pass
            adjusted_width = min(max_length + 2, 50)
            ws.column_dimensions[column_letter].width = adjusted_width
 
    wb.save(file_path)
    print(f"✅ Formatting applied: {file_path}")
 
# Apply formatting
format_excel_report(report_file)

# Quick check: how many users registered in a week?
df = pd.read_sql("SELECT * FROM users WHERE created_at >= NOW() - INTERVAL '7 days'", engine)
print(df.groupby(df['created_at'].dt.date).size())

# Combining data from DB and CSV
users_db = pd.read_sql("SELECT * FROM users", engine)
purchases_csv = pd.read_csv('external_purchases.csv')
merged = users_db.merge(purchases_csv, on='user_id')

# Pivot table, aggregation by multiple dimensions
pivot = df.pivot_table(
    values='revenue',
    index='date',
    columns=['city', 'category'],
    aggfunc=['sum', 'mean']
)

df.to_excel('report.xlsx', sheet_name='Sales')

# Writing to DB in production
# ❌ df.to_sql('users', engine)  # Slow for large data
# ✅ Use bulk_insert_mappings or COPY

# ACID required, rollback, constraints
with session.begin():
    session.add(user)
    session.add(order)

# Processing 10 million rows — use SQL + indexes
# pandas will load everything into memory and may crash

# Use SQL for filtering, pandas for analysis
query = """
    SELECT * FROM orders
    WHERE created_at >= '2025-01-01'
    AND status = 'completed'
    LIMIT 100000
"""
df = pd.read_sql(query, engine)  # SQL: filtering
top_products = df.groupby('product_id')['quantity'].sum().nlargest(10)  # pandas: aggregation

import pandas as pd
 
df = pd.read_csv('data.csv')
result = (
    df.groupby('category')['sales']
    .sum()
    .sort_values(ascending=False)
    .head(10)
)

import polars as pl
 
# 5-10x faster than pandas!
df = pl.read_csv('data.csv')
result = (
    df.lazy()  # Lazy execution
    .group_by('category')
    .agg(pl.col('sales').sum())
    .sort('sales', descending=True)
    .limit(10)
    .collect()  # Execution
)

import duckdb
 
# SQL syntax, works with pandas DataFrame
df = pd.read_csv('data.csv')
result = duckdb.sql("""
    SELECT category, SUM(sales) as total_sales
    FROM df
    GROUP BY category
    ORDER BY total_sales DESC
    LIMIT 10
""").to_df()

# ✅ Parquet (snappy) — golden mean for most tasks
df.to_parquet('data.parquet', compression='snappy', index=False)
df = pd.read_parquet('data.parquet')
 
# ✅ CSV — only for exchange with external systems
df.to_csv('export.csv', index=False)
 
# ❌ Pickle — DON'T use for long-term storage
# Reasons: incompatible between pandas versions, security vulnerability

# ❌ Bad: everything loads as object (strings)
df = pd.read_csv('large_file.csv')
 
# ✅ Good: specify types explicitly
df = pd.read_csv(
    'large_file.csv',
    dtype={
        'user_id': 'int32',  # int32 instead of int64 if values < 2 billion
        'category': 'category',  # Saves memory for repeating strings
        'price': 'float32'
    },
    parse_dates=['created_at']
)
 
# Check memory usage
print(df.memory_usage(deep=True))
 
# Memory saving example
# int64 → int32: 50% memory saved (8 bytes → 4 bytes)
# object → category: 90% saved for repeating values

# ❌ Bad: loads entire file into memory
df = pd.read_csv('huge_file.csv')  # 10GB file — OOM
 
# ✅ Good: process in chunks
chunks = []
for chunk in pd.read_csv('huge_file.csv', chunksize=10000):
    # Process chunk
    filtered = chunk[chunk['status'] == 'active']
    chunks.append(filtered)
 
df = pd.concat(chunks, ignore_index=True)

# Slow: each row — Python function call
df['price_with_tax'] = df.apply(lambda row: row['price'] * 1.2, axis=1)
 
# What happens:
# for row in df.iterrows():  # Python loop
#     result = lambda(row)    # Python function call
#     # GIL blocks parallelism
#     # Interpreter overhead on each iteration

# Fast: operation over entire array at once
df['price_with_tax'] = df['price'] * 1.2
 
# What happens:
# Operation delegated to NumPy (written in C)
# Uses SIMD CPU instructions (parallel processing)
# No Python interpreter overhead
# Works 10-100x faster

import pandas as pd
import numpy as np
import time
 
# Create test DataFrame (100k rows)
df = pd.DataFrame({
    'price': np.random.randint(100, 10000, 100000),
    'quantity': np.random.randint(1, 100, 100000)
})
 
# Test 1: apply() with lambda
start = time.time()
df['total_apply'] = df.apply(lambda row: row['price'] * row['quantity'], axis=1)
print(f"apply(): {time.time() - start:.3f} sec")  # ~1.2 sec
 
# Test 2: Vectorized operation
start = time.time()
df['total_vector'] = df['price'] * df['quantity']
print(f"vectorized: {time.time() - start:.3f} sec")  # ~0.003 sec
 
# Difference: 400x faster! ⚡

# ✅ Acceptable: complex logic that can't be vectorized
def complex_transformation(row):
    # Logic with external API calls, regex, etc.
    if row['status'] == 'pending' and row['amount'] > threshold:
        return external_api_call(row['id'])
    return None
 
df['result'] = df.apply(complex_transformation, axis=1)
 
# But even here it's better:
# 1. First filter needed rows
# 2. Apply apply() only to them
filtered = df[(df['status'] == 'pending') & (df['amount'] > threshold)]
filtered['result'] = filtered.apply(complex_transformation, axis=1)

# ❌ Slow: apply for categorization
df['category'] = df.apply(
    lambda row: 'expensive' if row['price'] > 1000 else 'cheap',
    axis=1
)
 
# ✅ Fast: np.where (for 2 options)
df['category'] = np.where(df['price'] > 1000, 'expensive', 'cheap')
 
# ✅ Fast: np.select (for 3+ options)
conditions = [
    df['price'] > 5000,
    df['price'] > 1000,
    df['price'] > 500
]
choices = ['premium', 'expensive', 'medium']
df['category'] = np.select(conditions, choices, default='cheap')
 
# ✅ Fast: pd.cut (for ranges)
df['price_range'] = pd.cut(
    df['price'],
    bins=[0, 500, 1000, 5000, float('inf')],
    labels=['cheap', 'medium', 'expensive', 'premium']
)

# ❌ ERROR: modifying potential copy
subset = df[df['sales'] > 100]
subset['discount'] = 0.1  # ⚠️ SettingWithCopyWarning!
# Changes may not apply or apply to original
 
# ✅ CORRECT: explicit intent
# Method 1: modify original DataFrame
df.loc[df['sales'] > 100, 'discount'] = 0.1
 
# Method 2: create explicit copy
subset = df[df['sales'] > 100].copy()
subset['discount'] = 0.1

# ❌ BAD: 1GB memory for repeating strings
df = pd.read_csv('data.csv')
print(df['country'].memory_usage(deep=True))  # 1GB
 
# ✅ GOOD: 100MB memory with category
df['country'] = df['country'].astype('category')
print(df['country'].memory_usage(deep=True))  # 100MB
 
# Savings: 10x for columns with small number of unique values

# ❌ BAD: crashes on large files
df = pd.read_csv('huge_10gb.csv')  # OutOfMemoryError
 
# ✅ GOOD: process in chunks
result = []
for chunk in pd.read_csv('huge_10gb.csv', chunksize=50000):
    # Filter and process only needed
    filtered = chunk[chunk['status'] == 'active']
    result.append(filtered)
 
df = pd.concat(result, ignore_index=True)

# ❌ SLOW: pandas guesses types for each column
df = pd.read_csv('data.csv')  # 10 seconds
 
# ✅ FAST: explicitly specify types
dtypes = {
    'id': 'int32',
    'price': 'float32',
    'category': 'category',
    'created_at': 'str'  # Parse separately
}
df = pd.read_csv(
    'data.csv',
    dtype=dtypes,
    parse_dates=['created_at']
)  # 2 seconds
 
# Gain: 5x speed + less memory

# ❌ NIGHTMARE: 1000 rows = 5 seconds
total = 0
for index, row in df.iterrows():
    total += row['price'] * row['quantity']
 
# ✅ EXCELLENT: 1000 rows = 0.001 second
total = (df['price'] * df['quantity']).sum()
 
# Difference: 5000x faster!

# ❌ BAD: everything in one function
def process_sales(df):
    # 50 lines of unreadable code
    df['margin'] = df['price'] - df['cost']
    df['category'] = df.apply(lambda x: categorize(x), axis=1)
    df['bonus'] = df.apply(lambda x: calculate_bonus(x), axis=1)
    # ... 40 more lines
    return df
 
# ✅ GOOD: separation of concerns
class SalesProcessor:
    @staticmethod
    def calculate_margin(df):
        """Margin calculation."""
        return df['price'] - df['cost']
 
    @staticmethod
    def categorize_products(df):
        """Product categorization."""
        conditions = [df['margin'] > 100, df['margin'] > 50]
        choices = ['Premium', 'Standard']
        return np.select(conditions, choices, default='Budget')
 
    @staticmethod
    def calculate_bonuses(df):
        """Bonus calculation."""
        return np.where(df['category'] == 'Premium', df['margin'] * 0.1, 0)
 
    def process(self, df):
        """Main processing method."""
        df = df.copy()
        df['margin'] = self.calculate_margin(df)
        df['category'] = self.categorize_products(df)
        df['bonus'] = self.calculate_bonuses(df)
        return df

import pandas as pd
from typing import List, Dict, Optional
 
def load_sales_data(file_path: str) -> pd.DataFrame:
    """Loads sales data from CSV."""
 
    return pd.read_csv(file_path)
 
def filter_by_category(
    df: pd.DataFrame,
    categories: List[str]
) -> pd.DataFrame:
    """Filters data by categories."""
 
    return df[df['category'].isin(categories)]
 
def calculate_metrics(df: pd.DataFrame) -> Dict[str, float]:
    """Calculates metrics."""
 
    return {
        'total_sales': float(df['sales'].sum()),
        'avg_price': float(df['price'].mean()),
        'max_quantity': int(df['quantity'].max())
    }

def get_column(df: pd.DataFrame, column: str) -> pd.Series:
    """Returns column as Series."""
 
    return df[column]
 
def process_series(series: pd.Series) -> pd.Series:
    """Processes Series."""
 
    return series.fillna(0).astype(int)

def find_product(
    df: pd.DataFrame,
    product_id: int
) -> Optional[pd.Series]:
    """Finds product by ID."""
 
    result = df[df['id'] == product_id]
    if result.empty:
        return None
    return result.iloc[0]

pip install pandas-stubs

from pandas import DataFrame, Series
from typing import Literal
 
def aggregate_data(
    df: DataFrame,
    column: str,
    method: Literal['sum', 'mean', 'count']
) -> float:
    """Aggregates data with type checking."""
 
    if method == 'sum':
        return float(df[column].sum())
    elif method == 'mean':
        return float(df[column].mean())
    else:
        return float(df[column].count())