SQL in Applications

-- Python code (VULNERABLE!)
username = request.form['username']
password = request.form['password']

query = f"SELECT * FROM users WHERE username = '{username}' AND password = '{password}'"
cursor.execute(query)

-- Search vulnerability
search = "'; SELECT password FROM users WHERE '1'='1"
query = f"SELECT * FROM products WHERE name LIKE '%{search}%'"
-- Exposes password data!

-- ORDER BY injection
sort = "name; DROP TABLE products"
query = f"SELECT * FROM products ORDER BY {sort}"
-- Destroys table!

-- UNION attack
id = "1 UNION SELECT username, password, NULL FROM users"
query = f"SELECT id, name, price FROM products WHERE id = {id}"
-- Leaks user credentials!

# SAFE - Parameterized query
username = request.form['username']
password = request.form['password']

query = "SELECT * FROM users WHERE username = %s AND password = %s"
cursor.execute(query, (username, password))

# Even if attacker enters: admin' --
# Database treats it as literal string "admin' --"
# Query becomes: WHERE username = 'admin'' --' AND password = '...'
# No injection possible!

// SAFE - Parameterized query
const username = req.body.username;
const password = req.body.password;

const query = 'SELECT * FROM users WHERE username = ? AND password = ?';
connection.execute(query, [username, password], (err, results) => {
  // Safe from SQL injection
});

// SAFE - Prepared statement
String query = "SELECT * FROM users WHERE username = ? AND password = ?";
PreparedStatement stmt = connection.prepareStatement(query);
stmt.setString(1, username);
stmt.setString(2, password);
ResultSet rs = stmt.executeQuery();

// SAFE - Prepared statement
$query = "SELECT * FROM users WHERE username = :username AND password = :password";
$stmt = $pdo->prepare($query);
$stmt->execute(['username' => $username, 'password' => $password]);

// User selects sort column from dropdown
sort_column = request.form['sort']  // Could be: "name", "price", "created_at"

// This doesn't work!
query = "SELECT * FROM products ORDER BY ?"
cursor.execute(query, (sort_column,))
// Database treats ? as a string literal, not a column name

// SAFE - Validate against whitelist
allowed_columns = ['name', 'price', 'created_at']
sort_column = request.form['sort']

if sort_column not in allowed_columns:
    sort_column = 'name'  // Default

query = f"SELECT * FROM products ORDER BY {sort_column}"
cursor.execute(query)
// Safe because we validated the column name

// User-facing names mapped to safe SQL
sort_mapping = {
    'product_name': 'name',
    'product_price': 'price',
    'date_created': 'created_at'
}

user_sort = request.form['sort']
safe_column = sort_mapping.get(user_sort, 'name')

query = f"SELECT * FROM products ORDER BY {safe_column}"
cursor.execute(query)

// Every request opens a new connection
app.get('/products', (req, res) => {
  const connection = mysql.createConnection({...});  // Slow!
  connection.connect();
  connection.query('SELECT * FROM products', (err, results) => {
    connection.end();  // Close connection
    res.json(results);
  });
});

// 100 requests = 100 connection opens = 10-20 seconds wasted!

// Create pool once at startup
const pool = mysql.createPool({
  host: 'localhost',
  user: 'app_user',
  password: 'password',
  database: 'mydb',
  connectionLimit: 10  // Max 10 concurrent connections
});

// Each request reuses a connection from pool
app.get('/products', (req, res) => {
  pool.query('SELECT * FROM products', (err, results) => {
    // Connection automatically returned to pool
    res.json(results);
  });
});

// 100 requests = reuse 10 connections = milliseconds!

from sqlalchemy import create_engine
from sqlalchemy.pool import QueuePool

# Create engine with connection pool
engine = create_engine(
    'postgresql://user:pass@localhost/db',
    poolclass=QueuePool,
    pool_size=10,        # Normal connections
    max_overflow=20,     // Extra connections under load
    pool_timeout=30,     // Wait 30s for connection
    pool_recycle=3600    // Recycle connections after 1 hour
)

# Each query gets connection from pool
with engine.connect() as conn:
    result = conn.execute("SELECT * FROM products")
    # Connection returned to pool automatically

# Define model
class Product(Base):
    __tablename__ = 'products'
    id = Column(Integer, primary_key=True)
    name = Column(String(100))
    price = Column(Numeric(10, 2))

# Query using ORM (simple and safe)
products = session.query(Product).filter(
    Product.price > 50
).order_by(Product.name).all()

# Generated SQL:
# SELECT * FROM products WHERE price > 50 ORDER BY name

# More verbose but full control
query = "SELECT * FROM products WHERE price > %s ORDER BY name"
cursor.execute(query, (50,))
products = cursor.fetchall()

# Use ORM for simple operations
user = User.query.filter_by(email=email).first()
user.last_login = datetime.now()
session.commit()

# Use raw SQL for complex analytics
report_query = """
    SELECT
        DATE_TRUNC('month', order_date) AS month,
        COUNT(*) AS orders,
        SUM(total) AS revenue
    FROM orders
    WHERE order_date >= %s
    GROUP BY month
    ORDER BY month DESC
"""
results = session.execute(report_query, (start_date,))

// Python with proper error handling
try:
    cursor.execute(query, params)
    connection.commit()
    return {"status": "success"}

except psycopg2.IntegrityError as e:
    connection.rollback()
    if 'duplicate key' in str(e):
        return {"error": "Email already exists"}
    return {"error": "Data integrity error"}

except psycopg2.OperationalError as e:
    connection.rollback()
    logger.error(f"Database connection error: {e}")
    return {"error": "Database unavailable, please try again"}

except Exception as e:
    connection.rollback()
    logger.error(f"Unexpected error: {e}")
    return {"error": "Internal server error"}

finally:
    cursor.close()  // Always close cursor

// Bad: Fetches all columns (including large text fields)
SELECT * FROM articles

// Good: Only fetch needed columns
SELECT id, title, author, published_date FROM articles

// Bad: Fetch all products, paginate in application
SELECT * FROM products ORDER BY created_at DESC

// Good: Paginate in database
SELECT * FROM products ORDER BY created_at DESC LIMIT 20 OFFSET 0

-- If you frequently query by email
CREATE INDEX idx_users_email ON users(email);

-- If you frequently filter by status and sort by created_at
CREATE INDEX idx_orders_status_created ON orders(status, created_at DESC);

// Bad: Insert one row at a time
for product in products:
    cursor.execute("INSERT INTO products VALUES (%s, %s)", (product.name, product.price))
    // 1000 products = 1000 round trips!

// Good: Bulk insert
values = [(p.name, p.price) for p in products]
cursor.executemany("INSERT INTO products (name, price) VALUES (%s, %s)", values)
// 1000 products = 1 round trip!

// Use Redis/Memcached for hot data
def get_product(product_id):
    # Check cache first
    cached = redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)

    # Cache miss: fetch from database
    product = db.query("SELECT * FROM products WHERE id = %s", (product_id,))

    # Store in cache (expire after 1 hour)
    redis.setex(f"product:{product_id}", 3600, json.dumps(product))

    return product

// Python with context manager
def transfer_money(from_account, to_account, amount):
    try:
        with engine.begin() as conn:  # Auto-commit on success
            # Deduct from source
            conn.execute(
                "UPDATE accounts SET balance = balance - %s WHERE id = %s",
                (amount, from_account)
            )

            # Add to destination
            conn.execute(
                "UPDATE accounts SET balance = balance + %s WHERE id = %s",
                (amount, to_account)
            )

            # Log transaction
            conn.execute(
                "INSERT INTO transaction_log (from_id, to_id, amount) VALUES (%s, %s, %s)",
                (from_account, to_account, amount)
            )

        # Transaction committed automatically
        return {"status": "success"}

    except Exception as e:
        # Transaction rolled back automatically
        logger.error(f"Transfer failed: {e}")
        return {"status": "error", "message": "Transfer failed"}


// Node.js
async function transferMoney(fromAccount, toAccount, amount) {
    const connection = await pool.getConnection();

    try {
        await connection.beginTransaction();

        await connection.execute(
            'UPDATE accounts SET balance = balance - ? WHERE id = ?',
            [amount, fromAccount]
        );

        await connection.execute(
            'UPDATE accounts SET balance = balance + ? WHERE id = ?',
            [amount, toAccount]
        );

        await connection.commit();
        return { status: 'success' };

    } catch (error) {
        await connection.rollback();
        console.error('Transfer failed:', error);
        return { status: 'error', message: 'Transfer failed' };

    } finally {
        connection.release();  // Return to pool
    }
}

// Bad: Hardcoded credentials (committed to git!)
connection = mysql.connect(
    host='localhost',
    user='admin',
    password='supersecret123',  // EXPOSED!
    database='production'
)

// Good: Use environment variables
import os

connection = mysql.connect(
    host=os.getenv('DB_HOST'),
    user=os.getenv('DB_USER'),
    password=os.getenv('DB_PASSWORD'),
    database=os.getenv('DB_NAME')
)

// .env file (not committed to git):
DB_HOST=localhost
DB_USER=app_user
DB_PASSWORD=complex_password_here
DB_NAME=myapp_production

// config.py
class Config:
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')

class DevelopmentConfig(Config):
    DEBUG = True
    SQLALCHEMY_ECHO = True  # Log all SQL queries

class ProductionConfig(Config):
    DEBUG = False
    SQLALCHEMY_ECHO = False  # Don't log queries in production

config = {
    'development': DevelopmentConfig,
    'production': ProductionConfig
}

// Use:
app.config.from_object(config[os.getenv('APP_ENV', 'development')])