SQL 실전 패턴 — 앱 개발에서 자주 쓰는 쿼리

-- 기본 페이지네이션
SELECT * FROM posts
ORDER BY created_at DESC
LIMIT 20 OFFSET 40;  -- 3페이지 (0-indexed: 페이지2)

-- 커서 기반 (대용량 데이터에 효율적)
-- 이전 페이지 마지막 항목의 id를 커서로 사용
SELECT * FROM posts
WHERE id < 150   -- 커서
ORDER BY id DESC
LIMIT 20;

-- LIKE 기반 (간단하지만 인덱스 비효율)
SELECT * FROM posts
WHERE title LIKE '%리액트%' OR content LIKE '%리액트%';

-- PostgreSQL 전문 검색 (빠름)
SELECT title, ts_rank(search_vector, query) AS rank
FROM posts, to_tsquery('korean', '리액트') query
WHERE search_vector @@ query
ORDER BY rank DESC;

-- 검색 벡터 컬럼 추가
ALTER TABLE posts ADD COLUMN search_vector TSVECTOR;
UPDATE posts SET search_vector = to_tsvector('english', title || ' ' || content);
CREATE INDEX idx_posts_search ON posts USING GIN(search_vector);

-- 오늘/이번 주/이번 달 가입자 수
SELECT
    COUNT(*) FILTER (WHERE created_at >= CURRENT_DATE) AS today,
    COUNT(*) FILTER (WHERE created_at >= DATE_TRUNC('week', NOW())) AS this_week,
    COUNT(*) FILTER (WHERE created_at >= DATE_TRUNC('month', NOW())) AS this_month,
    COUNT(*) AS total
FROM users;

-- 일별 매출 추이 (최근 30일)
SELECT
    DATE(created_at) AS date,
    COUNT(*) AS order_count,
    SUM(amount) AS revenue
FROM orders
WHERE created_at >= NOW() - INTERVAL '30 days'
GROUP BY DATE(created_at)
ORDER BY date;

-- 부서별 급여 순위
SELECT
    name,
    department,
    salary,
    ROW_NUMBER() OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
    LAG(salary) OVER (PARTITION BY department ORDER BY salary DESC) AS prev_salary,
    salary - LAG(salary) OVER (PARTITION BY department ORDER BY salary DESC) AS diff
FROM employees;

-- 누적 합계 (Running Total)
SELECT
    date,
    revenue,
    SUM(revenue) OVER (ORDER BY date) AS cumulative_revenue
FROM daily_sales;

import psycopg2
from contextlib import contextmanager

DB_CONFIG = {
    "host": "localhost",
    "database": "mydb",
    "user": "postgres",
    "password": "password",
}

@contextmanager
def get_connection():
    conn = psycopg2.connect(**DB_CONFIG)
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

def get_users(page: int = 1, page_size: int = 20):
    with get_connection() as conn:
        with conn.cursor() as cur:
            # 파라미터화된 쿼리 (SQL 인젝션 방지)
            cur.execute(
                "SELECT id, name, email FROM users ORDER BY id LIMIT %s OFFSET %s",
                (page_size, (page - 1) * page_size)
            )
            columns = [d[0] for d in cur.description]
            return [dict(zip(columns, row)) for row in cur.fetchall()]

// schema.prisma
model User {
    id        Int      @id @default(autoincrement())
    name      String
    email     String   @unique
    posts     Post[]
    createdAt DateTime @default(now())
}

// 사용
import { PrismaClient } from "@prisma/client";
const prisma = new PrismaClient();

// 조회
const users = await prisma.user.findMany({
    where: { name: { contains: "철수" } },
    orderBy: { createdAt: "desc" },
    take: 20,
    skip: 0,
    include: { posts: { take: 3 } },
});

// 생성
const user = await prisma.user.create({
    data: { name: "철수", email: "kim@example.com" },
});

// 원시 SQL
const result = await prisma.$queryRaw`
    SELECT department, AVG(salary) as avg_salary
    FROM employees
    GROUP BY department
`;

# ❌ 위험: 직접 문자열 삽입
query = f"SELECT * FROM users WHERE name = '{user_input}'"
# user_input = "'; DROP TABLE users; --"  → 재앙!

# ✅ 안전: 파라미터화된 쿼리
cur.execute("SELECT * FROM users WHERE name = %s", (user_input,))