"""Type guard functions for runtime type checking in SQLSpec.
This module provides type-safe runtime checks that help the type checker
understand type narrowing, replacing defensive hasattr() and duck typing patterns.
"""
import inspect
import sys
from collections.abc import Sequence
from collections.abc import Set as AbstractSet
from dataclasses import Field
from dataclasses import fields as dataclasses_fields
from dataclasses import is_dataclass as dataclasses_is_dataclass
from typing import TYPE_CHECKING, Any, Literal, cast
from sqlglot import exp
from typing_extensions import is_typeddict
from sqlspec._typing import Empty
from sqlspec.protocols import (
DictProtocol,
HasExpressionAndParametersProtocol,
HasExpressionAndSQLProtocol,
HasExpressionProtocol,
HasMigrationConfigProtocol,
HasParameterBuilderProtocol,
HasSQLGlotExpressionProtocol,
HasStatementConfigFactoryProtocol,
HasValueProtocol,
SupportsArrowResults,
WithMethodProtocol,
)
from sqlspec.typing import (
ATTRS_INSTALLED,
LITESTAR_INSTALLED,
MSGSPEC_INSTALLED,
PYDANTIC_INSTALLED,
DataclassProtocol,
Struct,
)
from sqlspec.utils.text import camelize, kebabize, pascalize
if TYPE_CHECKING:
from typing import TypeGuard
from sqlspec._typing import AttrsInstanceStub, BaseModelStub, DTODataStub, StructStub
from sqlspec.core import StatementFilter
from sqlspec.core.parameters import TypedParameter
from sqlspec.protocols import (
ArrowTableStatsProtocol,
AsyncDeleteProtocol,
AsyncReadableProtocol,
AsyncReadBytesProtocol,
AsyncWriteBytesProtocol,
CursorMetadataProtocol,
HasAddListenerProtocol,
HasAsDictProtocol,
HasConfigProtocol,
HasConnectionConfigProtocol,
HasDatabaseUrlAndBindKeyProtocol,
HasErrorsProtocol,
HasExtensionConfigProtocol,
HasFieldNameProtocol,
HasFilterAttributesProtocol,
HasGetDataProtocol,
HasLastRowIdProtocol,
HasNameProtocol,
HasNotifiesProtocol,
HasRowcountProtocol,
HasSqliteErrorProtocol,
HasSqlStateProtocol,
HasStatementTypeProtocol,
HasTracerProviderProtocol,
HasTypeCodeProtocol,
HasTypecodeProtocol,
HasTypecodeSizedProtocol,
HasWhereProtocol,
MappingLikeProtocol,
NotificationProtocol,
PipelineCapableProtocol,
QueryResultProtocol,
ReadableProtocol,
SpanAttributeProtocol,
SupportsArrayProtocol,
SupportsCloseProtocol,
SupportsDtypeStrProtocol,
SupportsJsonTypeProtocol,
)
from sqlspec.typing import SupportedSchemaModel
__all__ = (
"dataclass_to_dict",
"expression_has_limit",
"extract_dataclass_fields",
"extract_dataclass_items",
"get_initial_expression",
"get_literal_parent",
"get_msgspec_rename_config",
"get_node_expressions",
"get_node_this",
"get_param_style_and_name",
"get_value_attribute",
"has_add_listener",
"has_array_interface",
"has_arrow_table_stats",
"has_asdict_method",
"has_config_attribute",
"has_connection_config",
"has_cursor_metadata",
"has_database_url_and_bind_key",
"has_dict_attribute",
"has_dtype_str",
"has_errors",
"has_expression_and_parameters",
"has_expression_and_sql",
"has_expression_attr",
"has_expressions_attribute",
"has_extension_config",
"has_field_name",
"has_filter_attributes",
"has_get_data",
"has_lastrowid",
"has_migration_config",
"has_name",
"has_notifies",
"has_parameter_builder",
"has_parent_attribute",
"has_pipeline_capability",
"has_query_result_metadata",
"has_rowcount",
"has_span_attribute",
"has_sqlglot_expression",
"has_sqlite_error",
"has_sqlstate",
"has_statement_config_factory",
"has_statement_type",
"has_this_attribute",
"has_tracer_provider",
"has_type_code",
"has_typecode",
"has_typecode_and_len",
"has_value_attribute",
"has_with_method",
"is_async_readable",
"is_attrs_instance",
"is_attrs_instance_with_field",
"is_attrs_instance_without_field",
"is_attrs_schema",
"is_copy_statement",
"is_dataclass",
"is_dataclass_instance",
"is_dataclass_with_field",
"is_dataclass_without_field",
"is_dict",
"is_dict_row",
"is_dict_with_field",
"is_dict_without_field",
"is_dto_data",
"is_expression",
"is_iterable_parameters",
"is_local_path",
"is_mapping_like",
"is_msgspec_struct",
"is_msgspec_struct_with_field",
"is_msgspec_struct_without_field",
"is_notification",
"is_number_literal",
"is_pydantic_model",
"is_pydantic_model_with_field",
"is_pydantic_model_without_field",
"is_readable",
"is_schema",
"is_schema_or_dict",
"is_schema_or_dict_with_field",
"is_schema_or_dict_without_field",
"is_schema_with_field",
"is_schema_without_field",
"is_statement_filter",
"is_string_literal",
"is_typed_dict",
"is_typed_parameter",
"resolve_row_format",
"supports_arrow_results",
"supports_async_delete",
"supports_async_read_bytes",
"supports_async_write_bytes",
"supports_close",
"supports_json_type",
"supports_where",
)
[docs]
def is_readable(obj: Any) -> "TypeGuard[ReadableProtocol]":
"""Check if an object is readable (has a read method)."""
try:
return callable(obj.read)
except AttributeError:
return False
[docs]
def is_async_readable(obj: Any) -> "TypeGuard[AsyncReadableProtocol]":
"""Check if an object exposes an async read method."""
try:
return callable(obj.read) and inspect.iscoroutinefunction(obj.read)
except AttributeError:
return False
[docs]
def is_notification(obj: Any) -> "TypeGuard[NotificationProtocol]":
"""Check if an object is a database notification with channel and payload."""
return hasattr(obj, "channel") and hasattr(obj, "payload")
[docs]
def has_pipeline_capability(obj: Any) -> "TypeGuard[PipelineCapableProtocol]":
"""Check if a connection supports pipeline execution."""
try:
return callable(obj.run_pipeline)
except AttributeError:
return False
[docs]
def has_array_interface(obj: Any) -> "TypeGuard[SupportsArrayProtocol]":
"""Check if an object supports the array interface (like NumPy arrays)."""
return hasattr(obj, "__array__")
[docs]
def has_add_listener(obj: Any) -> "TypeGuard[HasAddListenerProtocol]":
"""Check if an object exposes add_listener()."""
try:
return callable(obj.add_listener)
except AttributeError:
return False
[docs]
def has_notifies(obj: Any) -> "TypeGuard[HasNotifiesProtocol]":
"""Check if an object exposes notifies."""
return hasattr(obj, "notifies")
[docs]
def has_extension_config(obj: Any) -> "TypeGuard[HasExtensionConfigProtocol]":
"""Check if an object exposes extension_config mapping."""
return hasattr(obj, "extension_config")
[docs]
def has_config_attribute(obj: Any) -> "TypeGuard[HasConfigProtocol]":
"""Check if an object exposes config attribute."""
return hasattr(obj, "config")
[docs]
def has_connection_config(obj: Any) -> "TypeGuard[HasConnectionConfigProtocol]":
"""Check if an object exposes connection_config mapping."""
return hasattr(obj, "connection_config")
[docs]
def has_database_url_and_bind_key(obj: Any) -> "TypeGuard[HasDatabaseUrlAndBindKeyProtocol]":
"""Check if an object exposes database_url and bind_key."""
return hasattr(obj, "database_url") and hasattr(obj, "bind_key")
[docs]
def has_name(obj: Any) -> "TypeGuard[HasNameProtocol]":
"""Check if an object exposes __name__."""
return hasattr(obj, "__name__")
[docs]
def has_field_name(obj: Any) -> "TypeGuard[HasFieldNameProtocol]":
"""Check if an object exposes field_name attribute."""
return hasattr(obj, "field_name")
[docs]
def has_filter_attributes(obj: Any) -> "TypeGuard[HasFilterAttributesProtocol]":
"""Check if an object exposes filter attribute set."""
return hasattr(obj, "field_name") and hasattr(obj, "operation") and hasattr(obj, "value")
[docs]
def has_get_data(obj: Any) -> "TypeGuard[HasGetDataProtocol]":
"""Check if an object exposes get_data()."""
try:
return callable(obj.get_data)
except AttributeError:
return False
[docs]
def has_arrow_table_stats(obj: Any) -> "TypeGuard[ArrowTableStatsProtocol]":
"""Check if an object exposes Arrow row/byte stats."""
return hasattr(obj, "num_rows") and hasattr(obj, "nbytes")
[docs]
def has_rowcount(obj: Any) -> "TypeGuard[HasRowcountProtocol]":
"""Check if a cursor exposes rowcount metadata."""
return hasattr(obj, "rowcount")
[docs]
def has_lastrowid(obj: Any) -> "TypeGuard[HasLastRowIdProtocol]":
"""Check if a cursor exposes lastrowid metadata."""
return hasattr(obj, "lastrowid")
[docs]
def has_dtype_str(obj: Any) -> "TypeGuard[SupportsDtypeStrProtocol]":
"""Check if a dtype exposes string descriptor."""
return hasattr(obj, "str")
[docs]
def has_statement_type(obj: Any) -> "TypeGuard[HasStatementTypeProtocol]":
"""Check if a cursor exposes statement_type metadata."""
return hasattr(obj, "statement_type")
[docs]
def has_typecode(obj: Any) -> "TypeGuard[HasTypecodeProtocol]":
"""Check if an array-like object exposes typecode."""
return hasattr(obj, "typecode")
[docs]
def has_typecode_and_len(obj: Any) -> "TypeGuard[HasTypecodeSizedProtocol]":
"""Check if an array-like object exposes typecode and length."""
return hasattr(obj, "typecode") and hasattr(obj, "__len__")
[docs]
def has_type_code(obj: Any) -> "TypeGuard[HasTypeCodeProtocol]":
"""Check if an object exposes type_code."""
return hasattr(obj, "type_code")
[docs]
def has_sqlstate(obj: Any) -> "TypeGuard[HasSqlStateProtocol]":
"""Check if an exception exposes sqlstate."""
return hasattr(obj, "sqlstate")
[docs]
def has_sqlite_error(obj: Any) -> "TypeGuard[HasSqliteErrorProtocol]":
"""Check if an exception exposes sqlite error details."""
return hasattr(obj, "sqlite_errorcode")
[docs]
def has_value_attribute(obj: Any) -> "TypeGuard[HasValueProtocol]":
"""Check if an object exposes a value attribute."""
return hasattr(obj, "value")
[docs]
def has_errors(obj: Any) -> "TypeGuard[HasErrorsProtocol]":
"""Check if an exception exposes errors."""
return hasattr(obj, "errors")
[docs]
def has_span_attribute(obj: Any) -> "TypeGuard[SpanAttributeProtocol]":
"""Check if a span exposes set_attribute."""
try:
return callable(obj.set_attribute)
except AttributeError:
return False
[docs]
def has_tracer_provider(obj: Any) -> "TypeGuard[HasTracerProviderProtocol]":
"""Check if an object exposes get_tracer."""
try:
return callable(obj.get_tracer)
except AttributeError:
return False
[docs]
def supports_async_read_bytes(obj: Any) -> "TypeGuard[AsyncReadBytesProtocol]":
"""Check if backend supports async read_bytes."""
try:
return callable(obj.read_bytes_async)
except AttributeError:
return False
[docs]
def supports_async_write_bytes(obj: Any) -> "TypeGuard[AsyncWriteBytesProtocol]":
"""Check if backend supports async write_bytes."""
try:
return callable(obj.write_bytes_async)
except AttributeError:
return False
[docs]
def supports_json_type(obj: Any) -> "TypeGuard[SupportsJsonTypeProtocol]":
"""Check if an object exposes JSON type support."""
return hasattr(obj, "JSON")
[docs]
def supports_close(obj: Any) -> "TypeGuard[SupportsCloseProtocol]":
"""Check if an object exposes close()."""
try:
return callable(obj.close)
except AttributeError:
return False
[docs]
def supports_async_delete(obj: Any) -> "TypeGuard[AsyncDeleteProtocol]":
"""Check if backend supports async delete."""
try:
return callable(obj.delete_async)
except AttributeError:
return False
[docs]
def supports_where(obj: Any) -> "TypeGuard[HasWhereProtocol]":
"""Check if an SQL expression supports WHERE clauses."""
try:
return callable(obj.where)
except AttributeError:
return False
[docs]
def is_typed_dict(obj: Any) -> "TypeGuard[type]":
"""Check if an object is a TypedDict class.
Args:
obj: The object to check
Returns:
True if the object is a TypedDict class, False otherwise
"""
return is_typeddict(obj)
[docs]
def is_statement_filter(obj: Any) -> "TypeGuard[StatementFilter]":
"""Check if an object implements the StatementFilter protocol.
Args:
obj: The object to check
Returns:
True if the object is a StatementFilter, False otherwise
"""
return getattr(obj, "_is_statement_filter", False) is True
[docs]
def is_dict_row(row: Any) -> "TypeGuard[dict[str, Any]]":
"""Check if a row is a dictionary.
Args:
row: The row to check
Returns:
True if the row is a dictionary, False otherwise
"""
return type(row) is dict
[docs]
def is_mapping_like(obj: Any) -> "TypeGuard[MappingLikeProtocol]":
"""Check if an object can be converted to dict via dict() constructor.
This matches database row types like sqlite3.Row, asyncpg.Record, psycopg.Row
that have keys() method and support __getitem__ access.
Args:
obj: The object to check
Returns:
True if the object has keys() method and __getitem__, False otherwise
"""
try:
return callable(obj.keys)
except AttributeError:
return False
[docs]
def has_asdict_method(obj: Any) -> "TypeGuard[HasAsDictProtocol]":
"""Check if an object has _asdict() method.
Args:
obj: The object to check
Returns:
True if the object has _asdict() method, False otherwise
"""
try:
return callable(obj._asdict)
except AttributeError:
return False
[docs]
def is_iterable_parameters(parameters: Any) -> "TypeGuard[Sequence[Any]]":
"""Check if parameters are iterable (but not string or dict).
Args:
parameters: The parameters to check
Returns:
True if the parameters are iterable, False otherwise
"""
return isinstance(parameters, Sequence) and not isinstance(parameters, (str, bytes, dict))
[docs]
def has_with_method(obj: Any) -> "TypeGuard[WithMethodProtocol]":
"""Check if an object has a callable 'with_' method.
This is a more specific check than hasattr for SQLGlot expressions.
Args:
obj: The object to check
Returns:
True if the object has a callable with_ method, False otherwise
"""
return isinstance(obj, WithMethodProtocol)
[docs]
def is_dataclass_instance(obj: Any) -> "TypeGuard[DataclassProtocol]":
"""Check if an object is a dataclass instance.
Args:
obj: An object to check.
Returns:
True if the object is a dataclass instance.
"""
if isinstance(obj, type):
return False
return dataclasses_is_dataclass(obj)
[docs]
def is_dataclass(obj: Any) -> "TypeGuard[DataclassProtocol]":
"""Check if an object is a dataclass.
Args:
obj: Value to check.
Returns:
bool
"""
return dataclasses_is_dataclass(obj)
[docs]
def is_dataclass_with_field(obj: Any, field_name: str) -> "TypeGuard[DataclassProtocol]":
"""Check if an object is a dataclass and has a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_dataclass(obj):
return False
return any(field.name == field_name for field in dataclasses_fields(obj))
[docs]
def is_dataclass_without_field(obj: Any, field_name: str) -> "TypeGuard[DataclassProtocol]":
"""Check if an object is a dataclass and does not have a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_dataclass(obj):
return False
return all(field.name != field_name for field in dataclasses_fields(obj))
[docs]
def is_pydantic_model(obj: Any) -> "TypeGuard[BaseModelStub]":
"""Check if a value is a pydantic model class or instance.
Args:
obj: Value to check.
Returns:
bool
"""
if not PYDANTIC_INSTALLED:
return False
pydantic = sys.modules.get("pydantic")
if pydantic is None:
return False
base_model = pydantic.BaseModel
if isinstance(obj, type):
try:
return issubclass(obj, base_model)
except TypeError:
return False
return isinstance(obj, base_model)
[docs]
def is_pydantic_model_with_field(obj: Any, field_name: str) -> "TypeGuard[BaseModelStub]":
"""Check if a pydantic model has a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_pydantic_model(obj):
return False
try:
fields = obj.model_fields
except AttributeError:
try:
fields = obj.__fields__ # type: ignore[attr-defined]
except AttributeError:
return False
return field_name in fields
[docs]
def is_pydantic_model_without_field(obj: Any, field_name: str) -> "TypeGuard[BaseModelStub]":
"""Check if a pydantic model does not have a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_pydantic_model(obj):
return False
try:
fields = obj.model_fields
except AttributeError:
try:
fields = obj.__fields__ # type: ignore[attr-defined]
except AttributeError:
return True
return field_name not in fields
[docs]
def is_msgspec_struct(obj: Any) -> "TypeGuard[StructStub]":
"""Check if a value is a msgspec struct class or instance.
Args:
obj: Value to check.
Returns:
bool
"""
if not MSGSPEC_INSTALLED:
return False
if isinstance(obj, type):
try:
return issubclass(obj, Struct)
except TypeError:
return False
return isinstance(obj, Struct)
[docs]
def is_msgspec_struct_with_field(obj: Any, field_name: str) -> "TypeGuard[StructStub]":
"""Check if a msgspec struct has a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_msgspec_struct(obj):
return False
from msgspec import structs
struct_type = obj if isinstance(obj, type) else type(obj)
fields = structs.fields(cast("Any", struct_type))
return any(field.name == field_name for field in fields)
[docs]
def is_msgspec_struct_without_field(obj: Any, field_name: str) -> "TypeGuard[StructStub]":
"""Check if a msgspec struct does not have a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_msgspec_struct(obj):
return False
from msgspec import structs
struct_type = obj if isinstance(obj, type) else type(obj)
fields = structs.fields(cast("Any", struct_type))
return all(field.name != field_name for field in fields)
def _detect_rename_pattern(field_name: str, encode_name: str) -> "str | None":
"""Detect the rename pattern by comparing field name transformations.
Args:
field_name: Original field name
encode_name: Encoded field name
Returns:
The detected rename pattern ("camel", "kebab", "pascal") or None
"""
key = (field_name, encode_name)
if key in _MSGSPEC_RENAME_PATTERN_CACHE:
return _MSGSPEC_RENAME_PATTERN_CACHE[key]
result: str | None
if encode_name == camelize(field_name) and encode_name != field_name:
result = "camel"
elif encode_name == kebabize(field_name) and encode_name != field_name:
result = "kebab"
elif encode_name == pascalize(field_name) and encode_name != field_name:
result = "pascal"
else:
result = None
_MSGSPEC_RENAME_PATTERN_CACHE[key] = result
return result
_MSGSPEC_RENAME_CONFIG_CACHE: "dict[object, str | None]" = {}
_MSGSPEC_RENAME_PATTERN_CACHE: "dict[tuple[str, str], str | None]" = {}
[docs]
def get_msgspec_rename_config(schema_type: type) -> "str | None":
"""Extract msgspec rename configuration from a struct type.
Analyzes field name transformations to detect the msgspec rename pattern.
Since msgspec doesn't store the original rename parameter directly, we infer it
by comparing field names with their encode_name values.
Args:
schema_type: The msgspec struct type to inspect.
Returns:
The rename configuration value ("camel", "kebab", "pascal", etc.) if detected,
None if no rename configuration exists or if not a msgspec struct.
"""
if schema_type in _MSGSPEC_RENAME_CONFIG_CACHE:
return _MSGSPEC_RENAME_CONFIG_CACHE[schema_type]
if not MSGSPEC_INSTALLED or not is_msgspec_struct(schema_type):
_MSGSPEC_RENAME_CONFIG_CACHE[schema_type] = None
return None
from msgspec import structs
fields: tuple[Any, ...] = structs.fields(cast("Any", schema_type))
if not fields:
_MSGSPEC_RENAME_CONFIG_CACHE[schema_type] = None
return None
for field in fields:
if field.name != field.encode_name:
rename_config = _detect_rename_pattern(field.name, field.encode_name)
_MSGSPEC_RENAME_CONFIG_CACHE[schema_type] = rename_config
return rename_config
_MSGSPEC_RENAME_CONFIG_CACHE[schema_type] = None
return None
[docs]
def is_attrs_instance(obj: Any) -> "TypeGuard[AttrsInstanceStub]":
"""Check if a value is an attrs class instance.
Args:
obj: Value to check.
Returns:
bool
"""
if not ATTRS_INSTALLED:
return False
from sqlspec.utils.module_loader import import_optional_attr
attrs_has = import_optional_attr("attrs", "has")
return bool(attrs_has(obj.__class__))
[docs]
def is_attrs_schema(cls: Any) -> "TypeGuard[type[AttrsInstanceStub]]":
"""Check if a class type is an attrs schema.
Args:
cls: Class to check.
Returns:
bool
"""
if not ATTRS_INSTALLED:
return False
from sqlspec.utils.module_loader import import_optional_attr
attrs_has = import_optional_attr("attrs", "has")
return bool(attrs_has(cls))
[docs]
def is_attrs_instance_with_field(obj: Any, field_name: str) -> "TypeGuard[AttrsInstanceStub]":
"""Check if an attrs instance has a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_attrs_instance(obj):
return False
from sqlspec.utils.module_loader import import_optional_attr
attrs_fields = import_optional_attr("attrs", "fields")
return any(field.name == field_name for field in attrs_fields(obj.__class__))
[docs]
def is_attrs_instance_without_field(obj: Any, field_name: str) -> "TypeGuard[AttrsInstanceStub]":
"""Check if an attrs instance does not have a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
if not is_attrs_instance(obj):
return False
from sqlspec.utils.module_loader import import_optional_attr
attrs_fields = import_optional_attr("attrs", "fields")
return all(field.name != field_name for field in attrs_fields(obj.__class__))
[docs]
def is_dict(obj: Any) -> "TypeGuard[dict[str, Any]]":
"""Check if a value is a dictionary.
Args:
obj: Value to check.
Returns:
bool
"""
return type(obj) is dict
[docs]
def is_dict_with_field(obj: Any, field_name: str) -> "TypeGuard[dict[str, Any]]":
"""Check if a dictionary has a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return is_dict(obj) and field_name in obj
[docs]
def is_dict_without_field(obj: Any, field_name: str) -> "TypeGuard[dict[str, Any]]":
"""Check if a dictionary does not have a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return is_dict(obj) and field_name not in obj
[docs]
def is_schema(obj: Any) -> "TypeGuard[SupportedSchemaModel]":
"""Check if a value is a msgspec Struct, Pydantic model, attrs instance, or schema class.
Args:
obj: Value to check.
Returns:
bool
"""
return (
is_msgspec_struct(obj)
or is_pydantic_model(obj)
or is_attrs_instance(obj)
or is_attrs_schema(obj)
or is_dataclass(obj)
)
[docs]
def is_schema_or_dict(obj: Any) -> "TypeGuard[SupportedSchemaModel | dict[str, Any]]":
"""Check if a value is a msgspec Struct, Pydantic model, or dict.
Args:
obj: Value to check.
Returns:
bool
"""
return is_schema(obj) or is_dict(obj)
[docs]
def is_schema_with_field(obj: Any, field_name: str) -> "TypeGuard[SupportedSchemaModel]":
"""Check if a value is a msgspec Struct or Pydantic model with a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return is_msgspec_struct_with_field(obj, field_name) or is_pydantic_model_with_field(obj, field_name)
[docs]
def is_schema_without_field(obj: Any, field_name: str) -> "TypeGuard[SupportedSchemaModel]":
"""Check if a value is a msgspec Struct or Pydantic model without a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return not is_schema_with_field(obj, field_name)
[docs]
def is_schema_or_dict_with_field(obj: Any, field_name: str) -> "TypeGuard[SupportedSchemaModel | dict[str, Any]]":
"""Check if a value is a msgspec Struct, Pydantic model, or dict with a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return is_schema_with_field(obj, field_name) or is_dict_with_field(obj, field_name)
[docs]
def is_schema_or_dict_without_field(obj: Any, field_name: str) -> "TypeGuard[SupportedSchemaModel | dict[str, Any]]":
"""Check if a value is a msgspec Struct, Pydantic model, or dict without a specific field.
Args:
obj: Value to check.
field_name: Field name to check for.
Returns:
bool
"""
return not is_schema_or_dict_with_field(obj, field_name)
[docs]
def is_dto_data(v: Any) -> "TypeGuard[DTODataStub[Any]]":
"""Check if a value is a Litestar DTOData object.
Args:
v: Value to check.
Returns:
bool
"""
if not LITESTAR_INSTALLED:
return False
data_structures = sys.modules.get("litestar.dto.data_structures")
return data_structures is not None and isinstance(v, data_structures.DTOData)
[docs]
def is_expression(obj: Any) -> "TypeGuard[exp.Expr]":
"""Check if a value is a sqlglot Expression.
Args:
obj: Value to check.
Returns:
bool
"""
return isinstance(obj, exp.Expr)
[docs]
def has_dict_attribute(obj: Any) -> "TypeGuard[DictProtocol]":
"""Check if an object has a __dict__ attribute.
Args:
obj: Value to check.
Returns:
bool
"""
return isinstance(obj, DictProtocol)
[docs]
def dataclass_to_dict(
obj: "DataclassProtocol",
exclude_none: bool = False,
exclude_empty: bool = False,
convert_nested: bool = True,
exclude: "AbstractSet[str] | None" = None,
) -> "dict[str, Any]":
"""Convert a dataclass instance to a dictionary.
Args:
obj: A dataclass instance.
exclude_none: Whether to exclude None values.
exclude_empty: Whether to exclude Empty values.
convert_nested: Whether to recursively convert nested dataclasses.
exclude: An iterable of fields to exclude.
Returns:
A dictionary of key/value pairs.
"""
ret = {}
for field in extract_dataclass_fields(obj, exclude_none, exclude_empty, exclude=exclude):
value = object.__getattribute__(obj, field.name)
if is_dataclass_instance(value) and convert_nested:
ret[field.name] = dataclass_to_dict(value, exclude_none, exclude_empty)
else:
ret[field.name] = value
return cast("dict[str, Any]", ret)
[docs]
def get_node_this(node: "exp.Expr", default: Any | None = None) -> Any:
"""Safely get the 'this' attribute from a SQLGlot node.
Args:
node: The SQLGlot expression node
default: Default value if 'this' attribute doesn't exist
Returns:
The value of node.this or the default value
"""
try:
return node.this
except AttributeError:
return default
[docs]
def has_this_attribute(node: "exp.Expr") -> bool:
"""Check if a node has the 'this' attribute without using hasattr().
Args:
node: The SQLGlot expression node
Returns:
True if the node has a 'this' attribute, False otherwise
"""
try:
_ = node.this
except AttributeError:
return False
return True
[docs]
def get_node_expressions(node: "exp.Expression", default: Any | None = None) -> Any:
"""Safely get the 'expressions' attribute from a SQLGlot node.
Args:
node: The SQLGlot expression node
default: Default value if 'expressions' attribute doesn't exist
Returns:
The value of node.expressions or the default value
"""
try:
return node.expressions
except AttributeError:
return default
[docs]
def has_expressions_attribute(node: "exp.Expression") -> bool:
"""Check if a node has the 'expressions' attribute without using hasattr().
Args:
node: The SQLGlot expression node
Returns:
True if the node has an 'expressions' attribute, False otherwise
"""
try:
_ = node.expressions
except AttributeError:
return False
return True
[docs]
def get_literal_parent(literal: "exp.Expression", default: Any | None = None) -> Any:
"""Safely get the 'parent' attribute from a SQLGlot literal.
Args:
literal: The SQLGlot expression
default: Default value if 'parent' attribute doesn't exist
Returns:
The value of literal.parent or the default value
"""
try:
return literal.parent
except AttributeError:
return default
[docs]
def has_parent_attribute(literal: "exp.Expression") -> bool:
"""Check if a literal has the 'parent' attribute without using hasattr().
Args:
literal: The SQLGlot expression
Returns:
True if the literal has a 'parent' attribute, False otherwise
"""
try:
_ = literal.parent
except AttributeError:
return False
return True
[docs]
def is_string_literal(literal: "exp.Literal") -> bool:
"""Check if a literal is a string literal without using hasattr().
Args:
literal: The SQLGlot Literal expression
Returns:
True if the literal is a string, False otherwise
"""
try:
return bool(literal.is_string)
except AttributeError:
try:
return isinstance(literal.this, str)
except AttributeError:
return False
[docs]
def is_number_literal(literal: "exp.Literal") -> bool:
"""Check if a literal is a number literal without using hasattr().
Args:
literal: The SQLGlot Literal expression
Returns:
True if the literal is a number, False otherwise
"""
try:
return bool(literal.is_number)
except AttributeError:
try:
if literal.this is not None:
float(str(literal.this))
return True
except (AttributeError, ValueError, TypeError):
pass
return False
[docs]
def get_initial_expression(context: Any) -> "exp.Expr | None":
"""Safely get initial_expression from context.
Args:
context: SQL processing context
Returns:
The initial expression or None if not available
"""
try:
return context.initial_expression # type: ignore[no-any-return]
except AttributeError:
return None
[docs]
def expression_has_limit(expr: "exp.Expression | None") -> bool:
"""Check if an expression has a limit clause.
Args:
expr: SQLGlot expression to check
Returns:
True if expression has limit in args, False otherwise
"""
if expr is None:
return False
try:
return "limit" in expr.args
except AttributeError:
return False
[docs]
def get_value_attribute(obj: Any) -> Any:
"""Safely get the 'value' attribute from an object.
Args:
obj: Object to get value from
Returns:
The value attribute or the object itself if no value attribute
"""
if isinstance(obj, HasValueProtocol):
return obj.value
return obj
[docs]
def get_param_style_and_name(param: Any) -> "tuple[str | None, str | None]":
"""Safely get style and name attributes from a parameter.
Args:
param: Parameter object
Returns:
Tuple of (style, name) or (None, None) if attributes don't exist
"""
try:
style = param.style
name = param.name
except AttributeError:
return None, None
return style, name
[docs]
def is_copy_statement(expression: Any) -> "TypeGuard[exp.Expr]":
"""Check if the SQL expression is a PostgreSQL COPY statement.
Args:
expression: The SQL expression to check
Returns:
True if this is a COPY statement, False otherwise
"""
if expression is None:
return False
try:
copy_expr = exp.Copy
except AttributeError:
copy_expr = None
if copy_expr is not None and isinstance(expression, copy_expr):
return True
if isinstance(expression, (exp.Command, exp.Anonymous)):
sql_text = str(expression).strip().upper()
return sql_text.startswith("COPY ")
return False
[docs]
def is_typed_parameter(obj: Any) -> "TypeGuard[TypedParameter]":
"""Check if an object is a typed parameter.
Args:
obj: The object to check
Returns:
True if the object is a TypedParameter, False otherwise
"""
from sqlspec.core.parameters import TypedParameter
return isinstance(obj, TypedParameter)
[docs]
def has_expression_and_sql(obj: Any) -> "TypeGuard[HasExpressionAndSQLProtocol]":
"""Check if an object has both 'expression' and 'sql' attributes.
This is commonly used to identify SQL objects in the builder system.
Args:
obj: The object to check
Returns:
True if the object has both attributes, False otherwise
"""
if not hasattr(obj, "expression"):
return False
if hasattr(obj, "raw_sql"):
return True
return hasattr(obj, "sql")
[docs]
def has_expression_and_parameters(obj: Any) -> "TypeGuard[HasExpressionAndParametersProtocol]":
"""Check if an object has both 'expression' and 'parameters' attributes.
This is used to identify objects that contain both SQL expressions
and parameter mappings.
Args:
obj: The object to check
Returns:
True if the object has both attributes, False otherwise
"""
return isinstance(obj, HasExpressionAndParametersProtocol)
WINDOWS_DRIVE_PATTERN_LENGTH = 3
[docs]
def is_local_path(uri: str) -> bool:
r"""Check if URI represents a local filesystem path.
Detects local paths including:
- file:// URIs
- Absolute paths (Unix: /, Windows: C:\)
- Relative paths (., .., ~)
Args:
uri: URI or path string to check.
Returns:
True if uri is a local path, False for remote URIs.
"""
if not uri:
return False
if "://" in uri and not uri.startswith("file://"):
return False
if uri.startswith("file://"):
return True
if uri.startswith("/"):
return True
if uri.startswith((".", "~")):
return True
if len(uri) >= WINDOWS_DRIVE_PATTERN_LENGTH and uri[1:3] == ":\\":
return True
return "/" in uri or "\\" in uri
[docs]
def supports_arrow_results(obj: Any) -> "TypeGuard[SupportsArrowResults]":
"""Check if object supports Arrow result format.
Use this type guard to check if a driver or adapter supports returning
query results in Apache Arrow format via select_to_arrow() method.
Args:
obj: Object to check for Arrow results support.
Returns:
True if object implements SupportsArrowResults protocol.
"""
return isinstance(obj, SupportsArrowResults)
[docs]
def has_parameter_builder(obj: Any) -> "TypeGuard[HasParameterBuilderProtocol]":
"""Check if an object has an add_parameter method."""
return isinstance(obj, HasParameterBuilderProtocol)
[docs]
def has_expression_attr(obj: Any) -> "TypeGuard[HasExpressionProtocol]":
"""Check if an object has an _expression attribute."""
return isinstance(obj, HasExpressionProtocol)
[docs]
def has_sqlglot_expression(obj: Any) -> "TypeGuard[HasSQLGlotExpressionProtocol]":
"""Check if an object has a sqlglot_expression property."""
return isinstance(obj, HasSQLGlotExpressionProtocol)
[docs]
def has_statement_config_factory(obj: Any) -> "TypeGuard[HasStatementConfigFactoryProtocol]":
"""Check if an object has a _create_statement_config method.
Used to check if a config object can create statement configs dynamically.
Args:
obj: The object to check.
Returns:
True if the object has a _create_statement_config method.
"""
return isinstance(obj, HasStatementConfigFactoryProtocol)
[docs]
def has_migration_config(obj: Any) -> "TypeGuard[HasMigrationConfigProtocol]":
"""Check if an object has a migration_config attribute.
Used to check if a database config supports migrations.
Args:
obj: The object to check.
Returns:
True if the object has a migration_config attribute.
"""
return isinstance(obj, HasMigrationConfigProtocol)
