Generate SQLAlchemy and Raw SQL Queries with AI

# Top 10 Customers Query Solutions

## 1. Schema Assumptions

- `orders` table: `id` (PK), `customer_id` (FK → customers.id), `total_amount` (NUMERIC/DECIMAL), `created_at` (TIMESTAMP), `status` (VARCHAR)
- `customers` table: `id` (PK), `name` (VARCHAR), `email` (VARCHAR), `country` (VARCHAR)
- Country values stored as exact strings `'Germany'` and `'France'`
- Status value `'completed'` is lowercase
- PostgreSQL connection string format assumed

---

## 2. SQLAlchemy ORM Version

```python
from datetime import datetime, timedelta
from decimal import Decimal
from typing import List, Tuple

from sqlalchemy import (
    Column, Integer, String, Numeric, DateTime, ForeignKey, create_engine, func
)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Session

Base = declarative_base()


class Customer(Base):
    __tablename__ = "customers"
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False)
    country = Column(String, nullable=False)
    orders = relationship("Order", back_populates="customer")


class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customers.id"), nullable=False)
    total_amount = Column(Numeric(12, 2), nullable=False)
    created_at = Column(DateTime, nullable=False)
    status = Column(String, nullable=False)
    customer = relationship("Customer", back_populates="orders")


def get_top_customers_orm(
    db_url: str,
    countries: Tuple[str, ...] = ("Germany", "France"),
    days: int = 90,
    limit: int = 10,
) -> List[Tuple[str, str, Decimal]]:
    engine = create_engine(db_url)
    SessionLocal = sessionmaker(bind=engine, expire_on_commit=False)

    cutoff = datetime.utcnow() - timedelta(days=days)
    total_spent = func.sum(Order.total_amount).label("total_spent")

    with SessionLocal() as session:  # type: Session
        results = (
            session.query(Customer.name, Customer.email, total_spent)
            .join(Order, Order.customer_id == Customer.id)
            .filter(
                Customer.country.in_(countries),
                Order.status == "completed",
                Order.created_at >= cutoff,
            )
            .group_by(Customer.id, Customer.name, Customer.email)
            .order_by(total_spent.desc())
            .limit(limit)
            .all()
        )
    return [(name, email, total) for name, email, total in results]


if __name__ == "__main__":
    rows = get_top_customers_orm("postgresql+psycopg2://user:pass@localhost/mydb")
    for name, email, total in rows:
        print(f"{name} <{email}>: {total}")
```

---

## 3. SQLAlchemy Core Version

```python
from datetime import datetime, timedelta
from decimal import Decimal
from typing import List, Tuple

from sqlalchemy import (
    MetaData, Table, Column, Integer, String, Numeric, DateTime, ForeignKey,
    create_engine, select, func
)

metadata = MetaData()

customers = Table(
    "customers", metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False),
    Column("email", String, nullable=False),
    Column("country", String, nullable=False),
)

orders = Table(
    "orders", metadata,
    Column("id", Integer, primary_key=True),
    Column("customer_id", Integer, ForeignKey("customers.id"), nullable=False),
    Column("total_amount", Numeric(12, 2), nullable=False),
    Column("created_at", DateTime, nullable=False),
    Column("status", String, nullable=False),
)


def get_top_customers_core(
    db_url: str,
    countries: Tuple[str, ...] = ("Germany", "France"),
    days: int = 90,
    limit: int = 10,
) -> List[Tuple[str, str, Decimal]]:
    engine = create_engine(db_url)
    cutoff = datetime.utcnow() - timedelta(days=days)
    total_spent = func.sum(orders.c.total_amount).label("total_spent")

    stmt = (
        select(customers.c.name, customers.c.email, total_spent)
        .select_from(customers.join(orders, orders.c.customer_id == customers.c.id))
        .where(
            customers.c.country.in_(countries),
            orders.c.status == "completed",
            orders.c.created_at >= cutoff,
        )
        .group_by(customers.c.id, customers.c.name, customers.c.email)
        .order_by(total_spent.desc())
        .limit(limit)
    )

    with engine.connect() as conn:
        result = conn.execute(stmt)
        return [tuple(row) for row in result.all()]


if __name__ == "__main__":
    for row in get_top_customers_core("postgresql+psycopg2://user:pass@localhost/mydb"):
        print(row)
```

---

## 4. Raw SQL Version

```python
from datetime import datetime, timedelta
from decimal import Decimal
from typing import List, Tuple

from sqlalchemy import create_engine, text

SQL_TOP_CUSTOMERS = """
SELECT
    c.name,
    c.email,
    SUM(o.total_amount) AS total_spent
FROM customers AS c
JOIN orders AS o ON o.customer_id = c.id
WHERE c.country = ANY(:countries)
  AND o.status = :status
  AND o.created_at >= :cutoff

# Top 10 Customers by Spending Query

## 1. Schema Assumptions

- `orders` table: `id` (PK), `customer_id` (FK), `total_amount` (DECIMAL/NUMERIC), `created_at` (TIMESTAMP), `status` (VARCHAR)
- `customers` table: `id` (PK), `name` (VARCHAR), `email` (VARCHAR), `country` (VARCHAR)
- `created_at` is stored in UTC or database timezone
- `status` is a string column with value `'completed'`
- Both tables have standard integer primary keys

## 2. SQLAlchemy ORM Version

```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import create_engine, func
from sqlalchemy.orm import declarative_base, Session
from sqlalchemy.sql import and_

# Model definitions
Base = declarative_base()

from sqlalchemy import Column, Integer, String, Numeric, DateTime, ForeignKey
from sqlalchemy.orm import relationship

class Customer(Base):
    __tablename__ = 'customers'
    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    email = Column(String, nullable=False)
    country = Column(String, nullable=False)
    orders = relationship('Order', back_populates='customer')

class Order(Base):
    __tablename__ = 'orders'
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey('customers.id'), nullable=False)
    total_amount = Column(Numeric, nullable=False)
    created_at = Column(DateTime, nullable=False)
    status = Column(String, nullable=False)
    customer = relationship('Customer', back_populates='orders')

def get_top_customers_orm(db_url: str) -> List[Tuple[str, str, float]]:
    """
    Fetch top 10 customers by spending in last 90 days using SQLAlchemy ORM.
    Returns list of tuples: (name, email, total_spent)
    """
    engine = create_engine(db_url)
    
    # Calculate date 90 days ago
    ninety_days_ago = datetime.utcnow() - timedelta(days=90)
    
    with Session(engine) as session:
        results = session.query(
            Customer.name,
            Customer.email,
            func.sum(Order.total_amount).label('total_spent')
        ).join(
            Order, Customer.id == Order.customer_id
        ).filter(
            and_(
                Order.status == 'completed',
                Order.created_at >= ninety_days_ago,
                Customer.country.in_(['Germany', 'France'])
            )
        ).group_by(
            Customer.id,
            Customer.name,
            Customer.email
        ).order_by(
            func.sum(Order.total_amount).desc()
        ).limit(10).all()
    
    return [(name, email, float(total)) for name, email, total in results]

if __name__ == '__main__':
    # Example usage
    db_url = 'postgresql://user:password@localhost:5432/dbname'
    top_customers = get_top_customers_orm(db_url)
    for name, email, total_spent in top_customers:
        print(f"{name} ({email}): ${total_spent:.2f}")
```

## 3. SQLAlchemy Core Version

```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import (
    create_engine, Table, Column, Integer, String, Numeric, DateTime,
    ForeignKey, select, func, and_, MetaData
)

def get_top_customers_core(db_url: str) -> List[Tuple[str, str, float]]:
    """
    Fetch top 10 customers by spending in last 90 days using SQLAlchemy Core.
    Returns list of tuples: (name, email, total_spent)
    """
    engine = create_engine(db_url)
    metadata = MetaData()
    
    # Define tables
    customers = Table(
        'customers',
        metadata,
        Column('id', Integer, primary_key=True),
        Column('name', String),
        Column('email', String),
        Column('country', String),
    )
    
    orders = Table(
        'orders',
        metadata,
        Column('id', Integer, primary_key=True),
        Column('customer_id', Integer, ForeignKey('customers.id')),
        Column('total_amount', Numeric),
        Column('created_at', DateTime),
        Column('status', String),
    )
    
    # Calculate date 90 days ago
    ninety_days_ago = datetime.utcnow() - timedelta(days=90)
    
    # Build query
    stmt = select(
        customers.c.name,
        customers.c.email,
        func.sum(orders.c.total_amount).label('total_spent')
    ).select_from(
        customers.join(orders, customers.c.id == orders.c.customer_id)
    ).where(
        and_(
            orders.c.status == 'completed',
            orders.c.created_at >= ninety_days_ago,
            customers.c.country.in_(['Germany', 'France'])
        )
    ).group_by(
        customers.c.id,
        customers.c.name,
        customers.c.email
    ).order_by(
        func.sum(orders.c.total_amount).desc()
    ).limit(10)
    
    with engine.connect() as conn:
        results = conn.execute(stmt).fetchall()
    
    return [(name, email, float(total)) for name, email, total in results]

if __name__ == '__main__':
    # Example usage
    db_url = 'postgresql://user:password@localhost:5432/dbname'
    top_customers = get_top_customers_core(db_url)
    for name, email, total_spent in top_customers:
        print(f"{name} ({email}): ${total_spent:.2f}")
```

## 4. Raw SQL Version

```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import create_engine, text

def get_top_customers_raw_sql(db_url: str) -> List[Tuple[str, str, float]]:
    """
    Fetch top 10 customers by spending in last 90 days using raw SQL.
    Returns list of tuples: (name, email, total_spent)
    """
    engine = create_engine(db_url)
    
    # Calculate date 90 days ago
    ninety_days_ago = datetime.utcnow() - timedelta(days=90)
    
    query = text("""
        SELECT
            c.name,
            c.email,
            SUM(o.total_amount) AS total_spent
        FROM customers c
        INNER JOIN orders o ON c.id = o.customer_id
        WHERE
            o.status = :status
            AND o.created_at >= :cutoff_date
            AND c.country IN (:country1, :country2)
        GROUP BY c.id, c.name, c.email
        ORDER BY total_spent DESC
        LIMIT :limit;
    """)
    
    with engine.connect() as conn:
        results = conn.execute(
            query,
            {
                'status': 'completed',
                'cutoff_date': ninety_days_ago,
                'country1': 'Germany',
                'country2': 'France',
                'limit': 10
            }
        ).fetchall()
    
    return [(name, email, float(total)) for name, email, total in results]

if __name__ == '__main__':
    # Example usage
    db_url = 'postgresql://user:password@localhost:5432/dbname'
    top_customers = get_top_customers_raw_sql(db_url)
    for name, email, total_spent in top_customers:
        print(f"{name} ({email}): ${total_spent:.2f}")
```

## 5. Explanation

- **Query Logic**: The query performs an

1. **Schema Assumptions**:
- `orders.id` and `customers.id` are primary keys.
- `orders.customer_id` is a foreign key to `customers.id`.
- `orders.total_amount` is a numeric/decimal column.
- `orders.created_at` is a timestamp/datetime column stored in PostgreSQL.
- `orders.status` contains values like `'completed'`.
- `customers.country` contains country names like `'Germany'` and `'France'`.

2. **SQLAlchemy ORM Version**:
```python
from __future__ import annotations

from decimal import Decimal
from typing import List, TypedDict

from sqlalchemy import DateTime, ForeignKey, Numeric, String, create_engine, func, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column, relationship


DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/your_database"


class Base(DeclarativeBase):
    pass


class Customer(Base):
    __tablename__ = "customers"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String, nullable=False)
    email: Mapped[str] = mapped_column(String, nullable=False)
    country: Mapped[str] = mapped_column(String, nullable=False)

    orders: Mapped[list["Order"]] = relationship(back_populates="customer")


class Order(Base):
    __tablename__ = "orders"

    id: Mapped[int] = mapped_column(primary_key=True)
    customer_id: Mapped[int] = mapped_column(ForeignKey("customers.id"), nullable=False)
    total_amount: Mapped[Decimal] = mapped_column(Numeric(12, 2), nullable=False)
    created_at: Mapped[DateTime] = mapped_column(DateTime(timezone=True), nullable=False)
    status: Mapped[str] = mapped_column(String, nullable=False)

    customer: Mapped[Customer] = relationship(back_populates="orders")


class CustomerSpendingResult(TypedDict):
    name: str
    email: str
    total_spent: Decimal


def get_top_customers_orm(database_url: str) -> List[CustomerSpendingResult]:
    engine = create_engine(database_url, future=True)

    countries = ["Germany", "France"]
    status = "completed"
    days_back = 90
    limit = 10

    stmt = (
        select(
            Customer.name,
            Customer.email,
            func.sum(Order.total_amount).label("total_spent"),
        )
        .join(Order, Order.customer_id == Customer.id)
        .where(Customer.country.in_(countries))
        .where(Order.status == status)
        .where(Order.created_at >= func.now() - func.make_interval(days=days_back))
        .group_by(Customer.id, Customer.name, Customer.email)
        .order_by(func.sum(Order.total_amount).desc())
        .limit(limit)
    )

    with Session(engine) as session:
        rows = session.execute(stmt).all()
        return [
            {
                "name": row.name,
                "email": row.email,
                "total_spent": row.total_spent,
            }
            for row in rows
        ]


if __name__ == "__main__":
    results = get_top_customers_orm(DATABASE_URL)
    for row in results:
        print(row)
```

3. **SQLAlchemy Core Version**:
```python
from __future__ import annotations

from decimal import Decimal
from typing import List, TypedDict

from sqlalchemy import (
    Column,
    DateTime,
    ForeignKey,
    Integer,
    MetaData,
    Numeric,
    String,
    Table,
    create_engine,
    func,
    select,
)


DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/your_database"

metadata = MetaData()

customers = Table(
    "customers",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String, nullable=False),
    Column("email", String, nullable=False),
    Column("country", String, nullable=False),
)

orders = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("customer_id", ForeignKey("customers.id"), nullable=False),
    Column("total_amount", Numeric(12, 2), nullable=False),
    Column("created_at", DateTime(timezone=True), nullable=False),
    Column("status", String, nullable=False),
)


class CustomerSpendingResult(TypedDict):
    name: str
    email: str
    total_spent: Decimal


def get_top_customers_core(database_url: str) -> List[CustomerSpendingResult]:
    engine = create_engine(database_url, future=True)

    countries = ["Germany", "France"]
    status = "completed"
    days_back = 90
    limit = 10

    total_spent = func.sum(orders.c.total_amount).label("total_spent")

    stmt = (
        select(customers.c.name, customers.c.email, total_spent)
        .select_from(
            customers.join(orders, orders.c.customer_id == customers.c.id)
        )
        .where(customers.c.country.in_(countries))
        .where(orders.c.status == status)
        .where(orders.c.created_at >= func.now() - func.make_interval(days=days_back))
        .group_by(customers.c.id, customers.c.name, customers.c.email)
        .order_by(total_spent.desc())
        .limit(limit)
    )

    with engine.connect() as conn:
        rows = conn.execute(stmt).mappings().all()
        return [
            {
                "name": row["name"],
                "email": row["email"],
                "total_spent": row["total_spent"],
            }
            for row in rows
        ]


if __name__ == "__main__":
    results = get_top_customers_core(DATABASE_URL)
    for row in results:
        print(row)
```

4. **Raw SQL Version**:
```python
from __future__ import annotations

from decimal import Decimal
from typing import List, TypedDict

from sqlalchemy import create_engine, text


DATABASE_URL = "postgresql+psycopg2://username:password@localhost:5432/your_database"


class CustomerSpendingResult(TypedDict):
    name: str
    email: str
    total_spent: Decimal


def get_top_customers_raw_sql(database_url: str) -> List[CustomerSpendingResult]:
    engine = create_engine(database_url, future=True)

    query = text(
        """
        SELECT
            c.name,
            c.email,
            SUM(o.total_amount) AS total_spent
        FROM customers AS c
        JOIN orders AS o
            ON o.customer_id = c.id
        WHERE c.country IN (:country1, :country2)
          AND o.status = :status
          AND o.created_at >= NOW() - make_interval(days => :days_back)
        GROUP BY c.id, c.name, c.email
        ORDER BY total_spent DESC
        LIMIT :limit_value
        """
    )

    params = {
        "country1": "Germany",
        "country2": "France",
        "status": "completed",
        "days_back": 90,
        "limit_value": 10,
    }

    with engine.connect() as conn:
        rows = conn.execute(query, params).mappings().all()
        return [
            {
                "name": row["name"],
                "email": row["email"],
                "total_spent": row["total_spent"],
            }
            for row in rows
        ]


if __name__ == "__main__":
    results = get_top_customers_raw_sql(DATABASE_URL)
    for row in results:
        print(row)
```

5. **Explanation**:
- The query joins `customers` and `orders`, filters to `country IN ('Germany', 'France')`, `status = 'completed'`, and `created_at` within the last 90 days, then aggregates with `SUM(total_amount)` per customer.
- Results are grouped by customer and sorted by total spending descending, returning only the top 10 customers.
- For performance, consider indexes such as `orders(customer_id)`, `orders(status, created_at)`, and `customers(country)`. Since aggregation is done in a single query, there is no N+1 query issue.

### 1. Schema Assumptions

*   `customers.id` is the primary key for the `customers` table.
*   `orders.id` is the primary key for the `orders` table.
*   `orders.customer_id` is a foreign key referencing `customers.id`.
*   `orders.total_amount` is a numeric type (e.g., `DECIMAL` or `FLOAT`).
*   `orders.created_at` is a timestamp or datetime type (e.g., `TIMESTAMP WITH TIME ZONE`).

---

### 2. SQLAlchemy ORM Version

```python
import os
from datetime import datetime, timedelta, timezone
from typing import List, Tuple, Any

from sqlalchemy import

1. **Schema Assumptions**:
   - `customers`: `id` (Integer PK), `name` (String), `email` (String), `country` (String).
   - `orders`: `id` (Integer PK), `customer_id` (Integer FK to `customers.id`), `total_amount` (Numeric), `created_at` (DateTime), `status` (String).
   - Indexes recommended: `orders(customer_id, created_at, status)`, `customers(country)`.

2. **SQLAlchemy ORM Version**:
```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import create_engine, Column, Integer, String, Numeric, DateTime, ForeignKey, func, desc
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from sqlalchemy.orm import Session

# Replace with your PostgreSQL connection string
engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname")
Base = declarative_base()

class Customer(Base):
    __tablename__ = "customers"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)
    country = Column(String)
    orders = relationship("Order", back_populates="customer")

class Order(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
    customer_id = Column(Integer, ForeignKey("customers.id"))
    total_amount = Column(Numeric(10, 2))
    created_at = Column(DateTime)
    status = Column(String)
    customer = relationship("Customer", back_populates="orders")

SessionLocal = sessionmaker(bind=engine)

date_filter: datetime = datetime.now() - timedelta(days=90)

with SessionLocal() as session:
    result: List[Tuple[str, str, float]] = (
        session.query(
            Customer.name,
            Customer.email,
            func.sum(Order.total_amount).label("total_spent"),
        )
        .join(Order)
        .filter(
            Customer.country.in_(["Germany", "France"]),
            Order.created_at >= date_filter,
            Order.status == "completed",
        )
        .group_by(Customer.id, Customer.name, Customer.email)
        .order_by(desc("total_spent"))
        .limit(10)
        .all()
    )
    print(result)  # List of (name, email, total_spent) tuples
```

3. **SQLAlchemy Core Version**:
```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import create_engine, Table, Column, Integer, String, Numeric, DateTime, MetaData, select, func, desc

# Replace with your PostgreSQL connection string
engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname")
metadata = MetaData()

customers = Table(
    "customers",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Column("email", String),
    Column("country", String),
)

orders = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("customer_id", Integer),
    Column("total_amount", Numeric(10, 2)),
    Column("created_at", DateTime),
    Column("status", String),
)

date_filter: datetime = datetime.now() - timedelta(days=90)
stmt = (
    select(
        customers.c.name,
        customers.c.email,
        func.sum(orders.c.total_amount).label("total_spent"),
    )
    .select_from(customers.join(orders, customers.c.id == orders.c.customer_id))
    .where(
        customers.c.country.in_(["Germany", "France"]),
        orders.c.created_at >= date_filter,
        orders.c.status == "completed",
    )
    .group_by(customers.c.id, customers.c.name, customers.c.email)
    .order_by(desc("total_spent"))
    .limit(10)
)

with engine.connect() as conn:
    result: List[Tuple[str, str, float]] = conn.execute(stmt).fetchall()
    print(result)  # List of (name, email, total_spent) tuples
```

4. **Raw SQL Version**:
```python
from datetime import datetime, timedelta
from typing import List, Tuple
from sqlalchemy import create_engine, text

# Replace with your PostgreSQL connection string
engine = create_engine("postgresql+psycopg2://user:pass@localhost/dbname")

date_filter: datetime = datetime.now() - timedelta(days=90)
query = text(
    """
    SELECT c.name, c.email, SUM(o.total_amount) AS total_spent
    FROM customers c
    JOIN orders o ON c.id = o.customer_id
    WHERE c.country IN ('Germany', 'France')
      AND o.created_at >= :date_filter
      AND o.status = 'completed'
    GROUP BY c.id, c.name, c.email
    ORDER BY total_spent DESC
    LIMIT 10
    """
)

with engine.connect() as conn:
    result: List[Tuple[str, str, float]] = conn.execute(query, {"date_filter": date_filter}).fetchall()
    print(result)  # List of (name, email, total_spent) tuples
```

5. **Explanation**:
   - Joins `customers` to `orders`, filters by `country IN ('Germany', 'France')`, `status='completed'`, and `created_at >= 90 days ago`; groups by customer identifiers, sums `total_amount`, sorts descending, limits to 10.
   - Parameterized `:date_filter` prevents injection; single query avoids N+1 issues.
   - Performance: Leverage indexes on `orders(created_at, status, customer_id)` and `customers(country)` for fast filtering/joins/grouping on potentially large tables.