Dynamic Filter Builder — Safe WHERE Clauses from User Input

Real applications need filters driven by user input: search forms, API query strings, dashboard dropdowns. Building these safely in Python without string concatenation requires accumulating clauses and params in parallel lists.

import psycopg2
from psycopg2.extras import RealDictCursor

def build_tweet_query(
    user_id: int | None = None,
    min_likes: int | None = None,
    hashtag: str | None = None,
    since: str | None = None,
    until: str | None = None,
    verified_only: bool = False,
    page: int = 1,
    page_size: int = 20,
) -> tuple[str, list]:
    """
    Build a safe, parameterised tweet search query from optional filters.
    Returns (sql_string, params_list).
    """
    clauses = ["t.retweet_of_id IS NULL"]   # always exclude retweets
    params  = []

    if user_id is not None:
        clauses.append("t.user_id = %s")
        params.append(user_id)

    if min_likes is not None:
        clauses.append("t.like_count >= %s")
        params.append(min_likes)

    if since is not None:
        clauses.append("t.created_at >= %s")
        params.append(since)

    if until is not None:
        clauses.append("t.created_at < %s")
        params.append(until)

    if verified_only:
        clauses.append("u.verified = TRUE")
        # no param needed — boolean literal is safe to interpolate

    # Hashtag join — only add if needed
    hashtag_join = ""
    if hashtag is not None:
        hashtag_join = """
            JOIN tw_tweet_hashtags th ON th.tweet_id = t.id
            JOIN tw_hashtags h ON h.id = th.hashtag_id
        """
        clauses.append("h.tag = %s")
        params.append(hashtag.lstrip("#").lower())

    # Pagination
    offset = (page - 1) * page_size
    params.extend([page_size, offset])

    sql = f"""
        SELECT t.id, u.username, t.content, t.like_count,
               t.retweet_count, t.created_at
        FROM tw_tweets t
        JOIN tw_users u ON u.id = t.user_id
        {hashtag_join}
        WHERE {" AND ".join(clauses)}
        ORDER BY t.like_count DESC
        LIMIT %s OFFSET %s
    """
    return sql, params


# ── Usage ─────────────────────────────────────────────────────────
conn = psycopg2.connect(host="127.0.0.1", dbname="shared_db",
    user="your_db_user", password="your_db_password")
cur = conn.cursor(cursor_factory=RealDictCursor)

# Search for high-engagement #sql tweets from verified users
sql, params = build_tweet_query(
    hashtag="#sql",
    min_likes=50,
    verified_only=True,
    since="2024-01-01",
    page=1,
    page_size=10,
)

print("Generated SQL:")
print(sql)
print("Params:", params)

cur.execute(sql, params)
for row in cur.fetchall():
    print(f"@{row['username']}: {row['like_count']} likes — {row['content'][:60]}")

conn.close()