Prisma Raw Queries — When and How to Use Custom SQL

import { PrismaClient } from '@prisma/client';

const prisma = new PrismaClient();

// $queryRaw: SELECT queries (returns rows)
const files = await prisma.$queryRaw`
  SELECT id, name, size_bytes, created_at
  FROM files
  WHERE size_bytes > ${minSize}
  ORDER BY created_at DESC
  LIMIT 10
`;

// $executeRaw: INSERT/UPDATE/DELETE (returns count)
const result = await prisma.$executeRaw`
  UPDATE files
  SET status = 'archived'
  WHERE created_at < ${cutoffDate}
  AND status = 'inactive'
`;

console.log(`Archived ${result} files`);

# raw_basics.py
# QueryRaw vs ExecuteRaw

def raw_comparison():
    print("$queryRaw vs $executeRaw:")
    print()
    print("$queryRaw")
    print("  - For SELECT queries")
    print("  - Returns array of rows")
    print("  - Can map to Prisma model types")
    print()
    print("$executeRaw")
    print("  - For INSERT, UPDATE, DELETE")
    print("  - Returns count of affected rows")
    print("  - Does NOT return the affected data")
    print()
    print("Both use tagged template literals:")
    print("  prisma.$queryRaw`SELECT * FROM users`")
    print("  prisma.$executeRaw`UPDATE users SET ...`")

raw_comparison()

// Safe parameterized queries
const email = 'user@example.com';
const minAge = 18;

// Use ${} for parameterized values
const user = await prisma.$queryRaw`
  SELECT * FROM users
  WHERE email = ${email}
  AND age >= ${minAge}
`;

// With typed output
type User = {
  id: number;
  email: string;
  name: string | null;
};

const users = await prisma.$queryRaw<User[]>`
  SELECT id, email, name
  FROM users
  WHERE status = 'active'
`;

// SQL injection attempt (safe with ${})
const maliciousInput = "'; DROP TABLE users; --";
const safeResult = await prisma.$queryRaw`
  SELECT * FROM users WHERE email = ${maliciousInput}
`;
// This safely escapes the input, no injection occurs

# parameterized.py
# Safe parameterized queries

def parameterized():
    print("Parameterized Queries:")
    print()
    print("Safe (use ${}):")
    print("  prisma.$queryRaw`SELECT * FROM users WHERE email = ${email}`")
    print("  - Prisma escapes the value")
    print("  - Prevents SQL injection")
    print("  - Type-safe")
    print()
    print("Unsafe (DON'T string interpolate):")
    print('  prisma.$queryRaw`SELECT * FROM users WHERE email = "${email}"`')
    print("  - Vulnerable to SQL injection")
    print("  - NEVER use string interpolation")
    print()
    print("For dynamic identifiers (table names, columns):")
    print("  Use Prisma.raw() or validate against allowlist")

parameterized()

// Define return type
interface FileSummary {
  id: string;
  name: string;
  sizeBytes: bigint;
  status: string;
  daysSinceUpload: number;
}

// Query with typed result
const summaries = await prisma.$queryRaw<FileSummary[]>`
  SELECT 
    id,
    name,
    size_bytes as "sizeBytes",
    status,
    EXTRACT(DAY FROM NOW() - created_at)::int as "daysSinceUpload"
  FROM files
  WHERE user_id = ${userId}
  ORDER BY created_at DESC
`;

// Using Prisma's built-in types
import { Prisma } from '@prisma/client';

const files = await prisma.$queryRaw<
  Prisma.FileGetPayload<true>[]
>`
  SELECT * FROM files WHERE size_bytes > ${minSize}
`;

# typed_raw.py
# Typed raw query results

def typed_results():
    print("Typed Raw Results:")
    print()
    print("Custom type:")
    print("  interface FileSummary {")
    print("    id: string;")
    print("    name: string;")
    print("    daysSinceUpload: number;")
    print("  }")
    print("  prisma.$queryRaw<FileSummary[]>`...`")
    print()
    print("Built-in Prisma types:")
    print("  import { Prisma } from '@prisma/client'")
    print("  prisma.$queryRaw<Prisma.FileGetPayload<true>[]>`...`")
    print()
    print("Column aliasing:")
    print("  SELECT name AS \"name\", created_at AS \"createdAt\"")
    print("  - Use quotes for camelCase aliases")

typed_results()

# when_raw.py
# When to use raw queries

def when_to_use_raw():
    print("Choose Raw SQL When:")
    print()
    print("1. Full-text search")
    print("   PostgreSQL: to_tsvector, to_tsquery")
    print()
    print("2. Window functions")
    print("   ROW_NUMBER(), RANK(), LAG(), LEAD()")
    print()
    print("3. Recursive CTEs")
    print("   WITH RECURSIVE for hierarchies")
    print()
    print("4. Complex aggregations")
    print("   Pivot tables, custom statistical functions")
    print()
    print("5. Bulk updates with complex conditions")
    print("   UPDATE with JOIN, subqueries")
    print()
    print("6. Database-specific features")
    print("   PostGIS, JSONB operators, array functions")
    print()
    print("Choose Prisma API When:")
    print("  - Standard CRUD operations")
    print("  - Filtering, sorting, pagination")
    print("  - Relations and nested writes")
    print("  - Transactions (raw queries can be included)")

when_to_use_raw()

def search_service():
    print("Full-Text Search Service:")
    print()
    print("Raw query:")
    print('  prisma.$queryRaw<SearchResult[]>`')
    print("    SELECT id, title, ts_rank(search_vector, query) AS rank")
    print("    FROM posts, plainto_tsquery('english', ${searchTerm}) query")
    print("    WHERE search_vector @@ query")
    print("    ORDER BY rank DESC")
    print("    LIMIT 20")
    print("  `")
    print()
    print("Typed result:")
    print("  interface SearchResult {")
    print("    id: number;")
    print("    title: string;")
    print("    rank: number;")
    print("  }")

search_service()