Source code for sqlspec.core.parameters._validator

"""Parameter extraction utilities."""

import re
from collections import OrderedDict
from typing import Final

from mypy_extensions import mypyc_attr

from sqlspec.core.parameters._types import ParameterInfo, ParameterStyle

__all__ = ("PARAMETER_REGEX", "ParameterValidator")

_PARAM_CHARS: Final[frozenset[str]] = frozenset("?%:@$")

PARAMETER_REGEX: Final[re.Pattern[str]] = re.compile(
    r"""
    (?P<dquote>"(?:[^"\\]|\\.)*") |
    (?P<squote>'(?:[^'\\]|\\.)*') |
    (?P<dollar_quoted_string>\$(?P<dollar_quote_tag_inner>\w*)?\$[\s\S]*?\$\4\$) |
    (?P<line_comment>--[^\r\n]*) |
    (?P<block_comment>/\*(?:[^*]|\*(?!/))*\*/) |
    (?P<pg_q_operator>\?\?|\?\||\?&) |
    (?P<pg_cast>::(?P<cast_type>\w+)) |
    (?P<sql_server_global>@@(?P<global_var_name>\w+)) |
    (?P<pyformat_named>%\((?P<pyformat_name>\w+)\)s) |
    (?P<pyformat_pos>%s) |
    (?P<positional_colon>(?<![A-Za-z0-9_]):(?P<colon_num>\d+)) |
    (?P<named_colon>(?<![A-Za-z0-9_]):(?P<colon_name>\w+)) |
    (?P<named_at>(?<![A-Za-z0-9_])@(?!sqlspec_)(?P<at_name>\w+)) |
    (?P<numeric>(?<![A-Za-z0-9_])\$(?P<numeric_num>\d+)) |
    (?P<named_dollar_param>(?<![A-Za-z0-9_])\$(?P<dollar_param_name>\w+)) |
    (?P<qmark>\?)
    """,
    re.VERBOSE | re.IGNORECASE | re.MULTILINE | re.DOTALL,
)

_SKIP_GROUPS: Final[tuple[str, ...]] = (
    "dquote",
    "squote",
    "dollar_quoted_string",
    "line_comment",
    "block_comment",
    "pg_q_operator",
    "pg_cast",
    "sql_server_global",
)



[docs]
@mypyc_attr(allow_interpreted_subclasses=False)
class ParameterValidator:
    """Extracts placeholder metadata and dialect compatibility information."""

    __slots__ = ("_cache_hits", "_cache_max_size", "_cache_misses", "_parameter_cache")


[docs]
    def __init__(self, cache_max_size: int = 5000) -> None:
        self._parameter_cache: OrderedDict[str, list[ParameterInfo]] = OrderedDict()
        self._cache_max_size = max(cache_max_size, 0)
        self._cache_hits = 0
        self._cache_misses = 0

[docs]

def set_cache_max_size(self, cache_max_size: int) -> None: """Update the maximum cache size for parameter metadata.""" self._cache_max_size = max(cache_max_size, 0) while len(self._parameter_cache) > self._cache_max_size: self._parameter_cache.popitem(last=False)