1"""
 2DDL generation helpers for convergence and schema management.
 3
 4Higher-level SQL builders that need database connections, ORM query machinery,
 5and constraint/index types. Separated from dialect.py (which is low-level and
 6imported everywhere) to avoid circular imports.
 7"""
 8
 9from __future__ import annotations
10
11from typing import TYPE_CHECKING, Any
12
13import psycopg.sql
14
15from plain.postgres.db import get_connection
16from plain.postgres.dialect import quote_name
17from plain.postgres.expressions import ExpressionList
18from plain.postgres.query_utils import Q
19from plain.postgres.sql.query import Query
20
21if TYPE_CHECKING:
22    from plain.postgres.base import Model
23    from plain.postgres.constraints import Deferrable
24
25
26def quote_value(value: Any) -> str:
27    """Quote a value for safe inclusion in a SQL string.
28
29    Not safe against injection from user code — intended only for SQL scripts,
30    default values, and constraint expressions (which are not user-defined).
31    """
32    if isinstance(value, str):
33        value = value.replace("%", "%%")
34    conn = get_connection()
35    return psycopg.sql.quote(value, conn.connection)
36
37
38def deferrable_sql(deferrable: Deferrable | None) -> str:
39    """Return the DEFERRABLE clause for a constraint, or empty string."""
40    from plain.postgres.constraints import (
41        Deferrable,  # circular: constraints imports ddl
42    )
43
44    if deferrable is None:
45        return ""
46    if deferrable == Deferrable.DEFERRED:
47        return " DEFERRABLE INITIALLY DEFERRED"
48    if deferrable == Deferrable.IMMEDIATE:
49        return " DEFERRABLE INITIALLY IMMEDIATE"
50    return ""
51
52
53def compile_expression_sql(model: type[Model], expression_q: Q) -> str:
54    """Compile a Q expression to a SQL string with quoted literal params."""
55    query = Query(model=model, alias_cols=False)
56    where = query.build_where(expression_q)
57    compiler = query.get_compiler()
58    conn = get_connection()
59    sql, params = where.as_sql(compiler, conn)
60    return sql % tuple(quote_value(p) for p in params)
61
62
63def compile_index_expressions_sql(
64    model: type[Model], expressions: tuple[Any, ...]
65) -> str:
66    """Compile index/constraint expressions (e.g. F(), OrderBy) to SQL."""
67    from plain.postgres.indexes import IndexExpression  # circular: indexes imports ddl
68
69    query = Query(model, alias_cols=False)
70    compiler = query.get_compiler()
71    index_expressions = [IndexExpression(expr) for expr in expressions]
72    expr_list = ExpressionList(*index_expressions).resolve_expression(query)
73    sql, params = compiler.compile(expr_list)
74    return sql % tuple(quote_value(p) for p in params)
75
76
77def build_include_sql(model: type[Model], include: tuple[str, ...]) -> str:
78    """Build the INCLUDE clause for an index or constraint, or empty string."""
79    if not include:
80        return ""
81    include_cols = [
82        quote_name(model._model_meta.get_forward_field(f).column) for f in include
83    ]
84    return " INCLUDE ({})".format(", ".join(include_cols))