Utility Functions#

SQLSpec exposes small, reusable helpers for runtime type narrowing, configuration, identifier generation, text and identifier formatting, serialization, schema conversion, correlation tracking, optional dependency loading, sync/async interoperation, deprecation, and fixture files. The modules and members documented on this page are supported public APIs; import helpers from their defining sqlspec.utils module.

Type Guards#

Runtime predicates in sqlspec.utils.type_guards narrow schema, mapping, SQLGlot expression, filter, cursor, and optional-library values.

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.

sqlspec.utils.type_guards.dataclass_to_dict(obj, exclude_none=False, exclude_empty=False, convert_nested=True, exclude=None)[source]#

Convert a dataclass instance to a dictionary.

Parameters:

  • obj (DataclassProtocol) -- A dataclass instance.

  • exclude_none (bool) -- Whether to exclude None values.

  • exclude_empty (bool) -- Whether to exclude Empty values.

  • convert_nested (bool) -- Whether to recursively convert nested dataclasses.

  • exclude (Set[str] | None) -- An iterable of fields to exclude.

Return type:

dict[str, typing.Any]

Returns:

A dictionary of key/value pairs.

sqlspec.utils.type_guards.expression_has_limit(expr)[source]#

Check if an expression has a limit clause.

Parameters:

expr (Expression | None) -- SQLGlot expression to check

Return type:

bool

Returns:

True if expression has limit in args, False otherwise

sqlspec.utils.type_guards.extract_dataclass_fields(obj, exclude_none=False, exclude_empty=False, include=None, exclude=None)[source]#

Extract dataclass fields.

Parameters:

  • obj (DataclassProtocol) -- A dataclass instance.

  • exclude_none (bool) -- Whether to exclude None values.

  • exclude_empty (bool) -- Whether to exclude Empty values.

  • include (Set[str] | None) -- An iterable of fields to include.

  • exclude (Set[str] | None) -- An iterable of fields to exclude.

Raises:

ValueError -- If there are fields that are both included and excluded.

Return type:

tuple[Field[typing.Any], ...]

Returns:

A tuple of dataclass fields.

sqlspec.utils.type_guards.extract_dataclass_items(obj, exclude_none=False, exclude_empty=False, include=None, exclude=None)[source]#

Extract name-value pairs from a dataclass instance.

Parameters:

  • obj (DataclassProtocol) -- A dataclass instance.

  • exclude_none (bool) -- Whether to exclude None values.

  • exclude_empty (bool) -- Whether to exclude Empty values.

  • include (Set[str] | None) -- An iterable of fields to include.

  • exclude (Set[str] | None) -- An iterable of fields to exclude.

Return type:

tuple[tuple[str, typing.Any], ...]

Returns:

A tuple of key/value pairs.

sqlspec.utils.type_guards.get_initial_expression(context)[source]#

Safely get initial_expression from context.

Parameters:

context (Any) -- SQL processing context

Return type:

Expr | None

Returns:

The initial expression or None if not available

sqlspec.utils.type_guards.get_literal_parent(literal, default=None)[source]#

Safely get the 'parent' attribute from a SQLGlot literal.

Parameters:

  • literal (Expression) -- The SQLGlot expression

  • default (Any | None) -- Default value if 'parent' attribute doesn't exist

Return type:

Any

Returns:

The value of literal.parent or the default value

sqlspec.utils.type_guards.get_msgspec_rename_config(schema_type)[source]#

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.

Parameters:

schema_type (type) -- The msgspec struct type to inspect.

Return type:

str | None

Returns:

The rename configuration value ("camel", "kebab", "pascal", etc.) if detected, None if no rename configuration exists or if not a msgspec struct.

sqlspec.utils.type_guards.get_node_expressions(node, default=None)[source]#

Safely get the 'expressions' attribute from a SQLGlot node.

Parameters:

  • node (Expression) -- The SQLGlot expression node

  • default (Any | None) -- Default value if 'expressions' attribute doesn't exist

Return type:

Any

Returns:

The value of node.expressions or the default value

sqlspec.utils.type_guards.get_node_this(node, default=None)[source]#

Safely get the 'this' attribute from a SQLGlot node.

Parameters:

  • node (Expr) -- The SQLGlot expression node

  • default (Any | None) -- Default value if 'this' attribute doesn't exist

Return type:

Any

Returns:

The value of node.this or the default value

sqlspec.utils.type_guards.get_param_style_and_name(param)[source]#

Safely get style and name attributes from a parameter.

Parameters:

param (Any) -- Parameter object

Return type:

tuple[str | None, str | None]

Returns:

Tuple of (style, name) or (None, None) if attributes don't exist

sqlspec.utils.type_guards.get_value_attribute(obj)[source]#

Safely get the 'value' attribute from an object.

Parameters:

obj (Any) -- Object to get value from

Return type:

Any

Returns:

The value attribute or the object itself if no value attribute

sqlspec.utils.type_guards.has_add_listener(obj)[source]#

Check if an object exposes add_listener().

Return type:

TypeGuard[HasAddListenerProtocol]

sqlspec.utils.type_guards.has_array_interface(obj)[source]#

Check if an object supports the array interface (like NumPy arrays).

Return type:

TypeGuard[SupportsArrayProtocol]

sqlspec.utils.type_guards.has_arrow_table_stats(obj)[source]#

Check if an object exposes Arrow row/byte stats.

Return type:

TypeGuard[ArrowTableStatsProtocol]

sqlspec.utils.type_guards.has_asdict_method(obj)[source]#

Check if an object has _asdict() method.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[HasAsDictProtocol]

Returns:

True if the object has _asdict() method, False otherwise

sqlspec.utils.type_guards.has_config_attribute(obj)[source]#

Check if an object exposes config attribute.

Return type:

TypeGuard[HasConfigProtocol]

sqlspec.utils.type_guards.has_connection_config(obj)[source]#

Check if an object exposes connection_config mapping.

Return type:

TypeGuard[HasConnectionConfigProtocol]

sqlspec.utils.type_guards.has_cursor_metadata(obj)[source]#

Check if an object has cursor metadata (description).

Return type:

TypeGuard[CursorMetadataProtocol]

sqlspec.utils.type_guards.has_database_url_and_bind_key(obj)[source]#

Check if an object exposes database_url and bind_key.

Return type:

TypeGuard[HasDatabaseUrlAndBindKeyProtocol]

sqlspec.utils.type_guards.has_dict_attribute(obj)[source]#

Check if an object has a __dict__ attribute.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[DictProtocol]

Returns:

bool

sqlspec.utils.type_guards.has_dtype_str(obj)[source]#

Check if a dtype exposes string descriptor.

Return type:

TypeGuard[SupportsDtypeStrProtocol]

sqlspec.utils.type_guards.has_errors(obj)[source]#

Check if an exception exposes errors.

Return type:

TypeGuard[HasErrorsProtocol]

sqlspec.utils.type_guards.has_expression_and_parameters(obj)[source]#

Check if an object has both 'expression' and 'parameters' attributes.

This is used to identify objects that contain both SQL expressions and parameter mappings.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[HasExpressionAndParametersProtocol]

Returns:

True if the object has both attributes, False otherwise

sqlspec.utils.type_guards.has_expression_and_sql(obj)[source]#

Check if an object has both 'expression' and 'sql' attributes.

This is commonly used to identify SQL objects in the builder system.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[HasExpressionAndSQLProtocol]

Returns:

True if the object has both attributes, False otherwise

sqlspec.utils.type_guards.has_expression_attr(obj)[source]#

Check if an object has an _expression attribute.

Return type:

TypeGuard[HasExpressionProtocol]

sqlspec.utils.type_guards.has_expressions_attribute(node)[source]#

Check if a node has the 'expressions' attribute without using hasattr().

Parameters:

node (Expression) -- The SQLGlot expression node

Return type:

bool

Returns:

True if the node has an 'expressions' attribute, False otherwise

sqlspec.utils.type_guards.has_extension_config(obj)[source]#

Check if an object exposes extension_config mapping.

Return type:

TypeGuard[HasExtensionConfigProtocol]

sqlspec.utils.type_guards.has_field_name(obj)[source]#

Check if an object exposes field_name attribute.

Return type:

TypeGuard[HasFieldNameProtocol]

sqlspec.utils.type_guards.has_filter_attributes(obj)[source]#

Check if an object exposes filter attribute set.

Return type:

TypeGuard[HasFilterAttributesProtocol]

sqlspec.utils.type_guards.has_get_data(obj)[source]#

Check if an object exposes get_data().

Return type:

TypeGuard[HasGetDataProtocol]

sqlspec.utils.type_guards.has_lastrowid(obj)[source]#

Check if a cursor exposes lastrowid metadata.

Return type:

TypeGuard[HasLastRowIdProtocol]

sqlspec.utils.type_guards.has_migration_config(obj)[source]#

Check if an object has a migration_config attribute.

Used to check if a database config supports migrations.

Parameters:

obj (Any) -- The object to check.

Return type:

TypeGuard[HasMigrationConfigProtocol]

Returns:

True if the object has a migration_config attribute.

sqlspec.utils.type_guards.has_name(obj)[source]#

Check if an object exposes __name__.

Return type:

TypeGuard[HasNameProtocol]

sqlspec.utils.type_guards.has_notifies(obj)[source]#

Check if an object exposes notifies.

Return type:

TypeGuard[HasNotifiesProtocol]

sqlspec.utils.type_guards.has_parameter_builder(obj)[source]#

Check if an object has an add_parameter method.

Return type:

TypeGuard[HasParameterBuilderProtocol]

sqlspec.utils.type_guards.has_parent_attribute(literal)[source]#

Check if a literal has the 'parent' attribute without using hasattr().

Parameters:

literal (Expression) -- The SQLGlot expression

Return type:

bool

Returns:

True if the literal has a 'parent' attribute, False otherwise

sqlspec.utils.type_guards.has_pipeline_capability(obj)[source]#

Check if a connection supports pipeline execution.

Return type:

TypeGuard[PipelineCapableProtocol]

sqlspec.utils.type_guards.has_query_result_metadata(obj)[source]#

Check if an object has query result metadata (tag/status).

Return type:

TypeGuard[QueryResultProtocol]

sqlspec.utils.type_guards.has_rowcount(obj)[source]#

Check if a cursor exposes rowcount metadata.

Return type:

TypeGuard[HasRowcountProtocol]

sqlspec.utils.type_guards.has_span_attribute(obj)[source]#

Check if a span exposes set_attribute.

Return type:

TypeGuard[SpanAttributeProtocol]

sqlspec.utils.type_guards.has_sqlglot_expression(obj)[source]#

Check if an object has a sqlglot_expression property.

Return type:

TypeGuard[HasSQLGlotExpressionProtocol]

sqlspec.utils.type_guards.has_sqlite_error(obj)[source]#

Check if an exception exposes sqlite error details.

Return type:

TypeGuard[HasSqliteErrorProtocol]

sqlspec.utils.type_guards.has_sqlstate(obj)[source]#

Check if an exception exposes sqlstate.

Return type:

TypeGuard[HasSqlStateProtocol]

sqlspec.utils.type_guards.has_statement_config_factory(obj)[source]#

Check if an object has a _create_statement_config method.

Used to check if a config object can create statement configs dynamically.

Parameters:

obj (Any) -- The object to check.

Return type:

TypeGuard[HasStatementConfigFactoryProtocol]

Returns:

True if the object has a _create_statement_config method.

sqlspec.utils.type_guards.has_statement_type(obj)[source]#

Check if a cursor exposes statement_type metadata.

Return type:

TypeGuard[HasStatementTypeProtocol]

sqlspec.utils.type_guards.has_this_attribute(node)[source]#

Check if a node has the 'this' attribute without using hasattr().

Parameters:

node (Expr) -- The SQLGlot expression node

Return type:

bool

Returns:

True if the node has a 'this' attribute, False otherwise

sqlspec.utils.type_guards.has_tracer_provider(obj)[source]#

Check if an object exposes get_tracer.

Return type:

TypeGuard[HasTracerProviderProtocol]

sqlspec.utils.type_guards.has_type_code(obj)[source]#

Check if an object exposes type_code.

Return type:

TypeGuard[HasTypeCodeProtocol]

sqlspec.utils.type_guards.has_typecode(obj)[source]#

Check if an array-like object exposes typecode.

Return type:

TypeGuard[HasTypecodeProtocol]

sqlspec.utils.type_guards.has_typecode_and_len(obj)[source]#

Check if an array-like object exposes typecode and length.

Return type:

TypeGuard[HasTypecodeSizedProtocol]

sqlspec.utils.type_guards.has_with_method(obj)[source]#

Check if an object has a callable with_ method.

This is a more specific check than hasattr for SQLGlot expressions.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[WithMethodProtocol]

Returns:

True if the object has a callable with_ method, False otherwise

sqlspec.utils.type_guards.is_async_readable(obj)[source]#

Check if an object exposes an async read method.

Return type:

TypeGuard[AsyncReadableProtocol]

sqlspec.utils.type_guards.is_attrs_instance(obj)[source]#

Check if a value is an attrs class instance.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[AttrsInstanceStub]

Returns:

bool

sqlspec.utils.type_guards.is_attrs_instance_with_field(obj, field_name)[source]#

Check if an attrs instance has a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[AttrsInstanceStub]

Returns:

bool

sqlspec.utils.type_guards.is_attrs_instance_without_field(obj, field_name)[source]#

Check if an attrs instance does not have a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[AttrsInstanceStub]

Returns:

bool

sqlspec.utils.type_guards.is_attrs_schema(cls)[source]#

Check if a class type is an attrs schema.

Parameters:

cls (Any) -- Class to check.

Return type:

TypeGuard[type[AttrsInstanceStub]]

Returns:

bool

sqlspec.utils.type_guards.is_copy_statement(expression)[source]#

Check if the SQL expression is a PostgreSQL COPY statement.

Parameters:

expression (Any) -- The SQL expression to check

Return type:

TypeGuard[Expr]

Returns:

True if this is a COPY statement, False otherwise

sqlspec.utils.type_guards.is_dataclass(obj)[source]#

Check if an object is a dataclass.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[DataclassProtocol]

Returns:

bool

sqlspec.utils.type_guards.is_dataclass_instance(obj)[source]#

Check if an object is a dataclass instance.

Parameters:

obj (Any) -- An object to check.

Return type:

TypeGuard[DataclassProtocol]

Returns:

True if the object is a dataclass instance.

sqlspec.utils.type_guards.is_dataclass_with_field(obj, field_name)[source]#

Check if an object is a dataclass and has a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DataclassProtocol]

Returns:

bool

sqlspec.utils.type_guards.is_dataclass_without_field(obj, field_name)[source]#

Check if an object is a dataclass and does not have a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DataclassProtocol]

Returns:

bool

sqlspec.utils.type_guards.is_dict(obj)[source]#

Check if a value is a dictionary.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_dict_row(row)[source]#

Check if a row is a dictionary.

Parameters:

row (Any) -- The row to check

Return type:

TypeGuard[dict[str, typing.Any]]

Returns:

True if the row is a dictionary, False otherwise

sqlspec.utils.type_guards.is_dict_with_field(obj, field_name)[source]#

Check if a dictionary has a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_dict_without_field(obj, field_name)[source]#

Check if a dictionary does not have a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_dto_data(v)[source]#

Check if a value is a Litestar DTOData object.

Parameters:

v (Any) -- Value to check.

Return type:

TypeGuard[DTODataStub[typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_expression(obj)[source]#

Check if a value is a sqlglot Expression.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[Expr]

Returns:

bool

sqlspec.utils.type_guards.is_iterable_parameters(parameters)[source]#

Check if parameters are iterable (but not string or dict).

Parameters:

parameters (Any) -- The parameters to check

Return type:

TypeGuard[Sequence[typing.Any]]

Returns:

True if the parameters are iterable, False otherwise

sqlspec.utils.type_guards.is_local_path(uri)[source]#

Check if URI represents a local filesystem path.

Detects local paths including:

  • file:// URIs

  • Absolute paths (Unix: /, Windows: C:)

  • Relative paths (., .., ~)

Parameters:

uri (str) -- URI or path string to check.

Return type:

bool

Returns:

True if uri is a local path, False for remote URIs.

sqlspec.utils.type_guards.is_mapping_like(obj)[source]#

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.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[MappingLikeProtocol]

Returns:

True if the object has keys() method and __getitem__, False otherwise

sqlspec.utils.type_guards.is_msgspec_struct(obj)[source]#

Check if a value is a msgspec struct class or instance.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[StructStub]

Returns:

bool

sqlspec.utils.type_guards.is_msgspec_struct_with_field(obj, field_name)[source]#

Check if a msgspec struct has a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[StructStub]

Returns:

bool

sqlspec.utils.type_guards.is_msgspec_struct_without_field(obj, field_name)[source]#

Check if a msgspec struct does not have a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[StructStub]

Returns:

bool

sqlspec.utils.type_guards.is_notification(obj)[source]#

Check if an object is a database notification with channel and payload.

Return type:

TypeGuard[NotificationProtocol]

sqlspec.utils.type_guards.is_number_literal(literal)[source]#

Check if a literal is a number literal without using hasattr().

Parameters:

literal (Literal) -- The SQLGlot Literal expression

Return type:

bool

Returns:

True if the literal is a number, False otherwise

sqlspec.utils.type_guards.is_pydantic_model(obj)[source]#

Check if a value is a pydantic model class or instance.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[BaseModelStub]

Returns:

bool

sqlspec.utils.type_guards.is_pydantic_model_with_field(obj, field_name)[source]#

Check if a pydantic model has a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[BaseModelStub]

Returns:

bool

sqlspec.utils.type_guards.is_pydantic_model_without_field(obj, field_name)[source]#

Check if a pydantic model does not have a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[BaseModelStub]

Returns:

bool

sqlspec.utils.type_guards.is_readable(obj)[source]#

Check if an object is readable (has a read method).

Return type:

TypeGuard[ReadableProtocol]

sqlspec.utils.type_guards.is_schema(obj)[source]#

Check if a value is a msgspec Struct, Pydantic model, attrs instance, or schema class.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]]

Returns:

bool

sqlspec.utils.type_guards.is_schema_or_dict(obj)[source]#

Check if a value is a msgspec Struct, Pydantic model, or dict.

Parameters:

obj (Any) -- Value to check.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any] | dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_schema_or_dict_with_field(obj, field_name)[source]#

Check if a value is a msgspec Struct, Pydantic model, or dict with a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any] | dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_schema_or_dict_without_field(obj, field_name)[source]#

Check if a value is a msgspec Struct, Pydantic model, or dict without a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any] | dict[str, typing.Any]]

Returns:

bool

sqlspec.utils.type_guards.is_schema_with_field(obj, field_name)[source]#

Check if a value is a msgspec Struct or Pydantic model with a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]]

Returns:

bool

sqlspec.utils.type_guards.is_schema_without_field(obj, field_name)[source]#

Check if a value is a msgspec Struct or Pydantic model without a specific field.

Parameters:

  • obj (Any) -- Value to check.

  • field_name (str) -- Field name to check for.

Return type:

TypeGuard[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]]

Returns:

bool

sqlspec.utils.type_guards.is_statement_filter(obj)[source]#

Check if an object implements the StatementFilter protocol.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[StatementFilter]

Returns:

True if the object is a StatementFilter, False otherwise

sqlspec.utils.type_guards.is_string_literal(literal)[source]#

Check if a literal is a string literal without using hasattr().

Parameters:

literal (Literal) -- The SQLGlot Literal expression

Return type:

bool

Returns:

True if the literal is a string, False otherwise

sqlspec.utils.type_guards.is_typed_dict(obj)[source]#

Check if an object is a TypedDict class.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[type]

Returns:

True if the object is a TypedDict class, False otherwise

sqlspec.utils.type_guards.is_typed_parameter(obj)[source]#

Check if an object is a typed parameter.

Parameters:

obj (Any) -- The object to check

Return type:

TypeGuard[TypedParameter]

Returns:

True if the object is a TypedParameter, False otherwise

sqlspec.utils.type_guards.resolve_row_format(rows, *, default='tuple')[source]#

Resolve the raw row format from the first row in a result set.

Return type:

Literal['dict', 'tuple', 'record']

sqlspec.utils.type_guards.supports_arrow_results(obj)[source]#

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.

Parameters:

obj (Any) -- Object to check for Arrow results support.

Return type:

TypeGuard[SupportsArrowResults]

Returns:

True if object implements SupportsArrowResults protocol.

sqlspec.utils.type_guards.supports_async_delete(obj)[source]#

Check if backend supports async delete.

Return type:

TypeGuard[AsyncDeleteProtocol]

sqlspec.utils.type_guards.supports_async_read_bytes(obj)[source]#

Check if backend supports async read_bytes.

Return type:

TypeGuard[AsyncReadBytesProtocol]

sqlspec.utils.type_guards.supports_async_write_bytes(obj)[source]#

Check if backend supports async write_bytes.

Return type:

TypeGuard[AsyncWriteBytesProtocol]

sqlspec.utils.type_guards.supports_close(obj)[source]#

Check if an object exposes close().

Return type:

TypeGuard[SupportsCloseProtocol]

sqlspec.utils.type_guards.supports_json_type(obj)[source]#

Check if an object exposes JSON type support.

Return type:

TypeGuard[SupportsJsonTypeProtocol]

sqlspec.utils.type_guards.supports_where(obj)[source]#

Check if an SQL expression supports WHERE clauses.

Return type:

TypeGuard[HasWhereProtocol]

Environment and Configuration#

Environment variable parsing utilities.

sqlspec.utils.env.get_config_val(key, default, type_hint=<sqlspec.utils.env._UnsetType object>)[source]#

Parse an environment variable value.

Overloads:

  • key (str), default (bool), type_hint (_UnsetType) → bool

  • key (str), default (int), type_hint (_UnsetType) → int

  • key (str), default (float), type_hint (_UnsetType) → float

  • key (str), default (str), type_hint (_UnsetType) → str

  • key (str), default (Path), type_hint (_UnsetType) → Path

  • key (str), default (list[Any]), type_hint (_UnsetType) → list[Any]

  • key (str), default (dict[str, Any]), type_hint (_UnsetType) → dict[str, Any]

  • key (str), default (None), type_hint (_UnsetType) → str | None

  • key (str), default (ParseType), type_hint (object) → Any

Parameters:
Returns:

Parsed value or the supplied default.

sqlspec.utils.env.get_config_val_with_aliases(key, aliases, default, type_hint=<sqlspec.utils.env._UnsetType object>)[source]#

Parse a canonical environment variable, falling back to aliases.

Overloads:

  • key (str), aliases (Sequence[str]), default (bool), type_hint (_UnsetType) → bool

  • key (str), aliases (Sequence[str]), default (int), type_hint (_UnsetType) → int

  • key (str), aliases (Sequence[str]), default (float), type_hint (_UnsetType) → float

  • key (str), aliases (Sequence[str]), default (str), type_hint (_UnsetType) → str

  • key (str), aliases (Sequence[str]), default (Path), type_hint (_UnsetType) → Path

  • key (str), aliases (Sequence[str]), default (list[Any]), type_hint (_UnsetType) → list[Any]

  • key (str), aliases (Sequence[str]), default (dict[str, Any]), type_hint (_UnsetType) → dict[str, Any]

  • key (str), aliases (Sequence[str]), default (None), type_hint (_UnsetType) → str | None

  • key (str), aliases (Sequence[str]), default (ParseType), type_hint (object) → Any

Parameters:

  • key (str) -- Canonical environment variable name.

  • aliases (Sequence[str]) -- Fallback variable names checked in order.

  • default (bool | int | float | str | Path | list[Any] | dict[str, Any] | None) -- Value returned when none of the variables are set.

  • type_hint (object) -- Optional parse target, including generic aliases such as list[int].

Returns:

Parsed value or the supplied default.

sqlspec.utils.env.get_env(key, default, type_hint=<sqlspec.utils.env._UnsetType object>)[source]#

Return a callable that parses an environment variable on demand.

Overloads:

  • key (str), default (bool), type_hint (_UnsetType) → Callable[[], bool]

  • key (str), default (int), type_hint (_UnsetType) → Callable[[], int]

  • key (str), default (float), type_hint (_UnsetType) → Callable[[], float]

  • key (str), default (str), type_hint (_UnsetType) → Callable[[], str]

  • key (str), default (Path), type_hint (_UnsetType) → Callable[[], Path]

  • key (str), default (list[Any]), type_hint (_UnsetType) → Callable[[], list[Any]]

  • key (str), default (dict[str, Any]), type_hint (_UnsetType) → Callable[[], dict[str, Any]]

  • key (str), default (None), type_hint (_UnsetType) → Callable[[], str | None]

  • key (str), default (ParseType), type_hint (object) → Callable[[], Any]

Parameters:
Returns:

Callable that returns the parsed value.

sqlspec.utils.env.get_env_with_aliases(key, aliases, default, type_hint=<sqlspec.utils.env._UnsetType object>)[source]#

Return a callable that parses a canonical environment key plus aliases.

Overloads:

  • key (str), aliases (Sequence[str]), default (bool), type_hint (_UnsetType) → Callable[[], bool]

  • key (str), aliases (Sequence[str]), default (int), type_hint (_UnsetType) → Callable[[], int]

  • key (str), aliases (Sequence[str]), default (float), type_hint (_UnsetType) → Callable[[], float]

  • key (str), aliases (Sequence[str]), default (str), type_hint (_UnsetType) → Callable[[], str]

  • key (str), aliases (Sequence[str]), default (Path), type_hint (_UnsetType) → Callable[[], Path]

  • key (str), aliases (Sequence[str]), default (list[Any]), type_hint (_UnsetType) → Callable[[], list[Any]]

  • key (str), aliases (Sequence[str]), default (dict[str, Any]), type_hint (_UnsetType) → Callable[[], dict[str, Any]]

  • key (str), aliases (Sequence[str]), default (None), type_hint (_UnsetType) → Callable[[], str | None]

  • key (str), aliases (Sequence[str]), default (ParseType), type_hint (object) → Callable[[], Any]

Parameters:

  • key (str) -- Canonical environment variable name.

  • aliases (Sequence[str]) -- Fallback variable names checked in order.

  • default (bool | int | float | str | Path | list[Any] | dict[str, Any] | None) -- Value returned when none of the variables are set.

  • type_hint (object) -- Optional parse target, including generic aliases such as list[int].

Returns:

Callable that returns the parsed value.

sqlspec.utils.env.is_env_set(key, aliases=())[source]#

Return whether a canonical environment variable or alias is present.

Parameters:

  • key (str) -- Canonical environment variable name.

  • aliases (Sequence[str]) -- Fallback variable names to check.

Return type:

bool

Returns:

True when any key is present in os.environ.

UUIDs and Compact Identifiers#

UUID and ID generation utilities with optional acceleration.

Provides wrapper functions for UUID construction, uuid3, uuid4, uuid5, uuid6, uuid7, and nanoid generation. Uses uuid-utils and fastnanoid packages for performance when available, falling back to the standard library.

When uuid-utils is installed:

  • uuid3, uuid4, uuid5, uuid6, uuid7 use the faster Rust implementation

  • uuid6 and uuid7 provide proper time-ordered UUIDs per RFC 9562

When uuid-utils is NOT installed:

  • uuid3, uuid4, uuid5 fall back silently to stdlib (equivalent output)

  • uuid6, uuid7 fall back to uuid4 with a warning (different UUID version)

When fastnanoid is installed:

  • nanoid() uses the Rust implementation for 21-char URL-safe IDs

When fastnanoid is NOT installed:

  • nanoid() falls back to uuid4().hex with a warning (different format)

sqlspec.utils.uuids.nanoid()[source]#

Generate a Nano ID.

Uses fastnanoid for performance when available. When fastnanoid is not installed, falls back to uuid4().hex with a warning.

Nano IDs are URL-safe, compact 21-character identifiers suitable for use as primary keys or short identifiers. The default alphabet uses A-Za-z0-9_- characters.

Return type:

str

Returns:

A 21-character Nano ID string, or 32-character UUID hex if fastnanoid unavailable.

sqlspec.utils.uuids.uuid3(name, namespace=None)[source]#

Generate a deterministic UUID (version 3) using MD5 hash.

Uses uuid-utils for performance when available, falls back to standard library uuid.uuid3() silently (equivalent output).

Parameters:

  • name (str) -- The name to hash within the namespace.

  • namespace (UUID | None) -- The namespace UUID. Defaults to NAMESPACE_DNS if not provided.

Return type:

UUID

Returns:

A deterministic UUID based on namespace and name.

sqlspec.utils.uuids.uuid4()[source]#

Generate a random UUID (version 4).

Uses uuid-utils for performance when available, falls back to standard library uuid.uuid4() silently (equivalent output).

Return type:

UUID

Returns:

A randomly generated UUID.

sqlspec.utils.uuids.uuid5(name, namespace=None)[source]#

Generate a deterministic UUID (version 5) using SHA-1 hash.

Uses uuid-utils for performance when available, falls back to standard library uuid.uuid5() silently (equivalent output).

Parameters:

  • name (str) -- The name to hash within the namespace.

  • namespace (UUID | None) -- The namespace UUID. Defaults to NAMESPACE_DNS if not provided.

Return type:

UUID

Returns:

A deterministic UUID based on namespace and name.

sqlspec.utils.uuids.uuid6()[source]#

Generate a time-ordered UUID (version 6).

Uses uuid-utils when available, falls back to Python 3.14+ native uuid.uuid6() or uuid4() with a warning.

UUIDv6 is lexicographically sortable by timestamp, making it suitable for database primary keys. It is a reordering of UUIDv1 fields to improve database performance.

Return type:

UUID

Returns:

A time-ordered UUID, or a random UUID if time-ordered generation unavailable.

sqlspec.utils.uuids.uuid7()[source]#

Generate a time-ordered UUID (version 7).

Uses uuid-utils when available, falls back to Python 3.14+ native uuid.uuid7() or uuid4() with a warning.

UUIDv7 is the recommended time-ordered UUID format per RFC 9562, providing millisecond precision timestamps. It is designed for modern distributed systems and database primary keys.

Return type:

UUID

Returns:

A time-ordered UUID, or a random UUID if time-ordered generation unavailable.

sqlspec.utils.uuids.uuid_from_bytes(value)[source]#

Construct a stdlib UUID from its 16-byte representation.

The stdlib constructor is retained for this shape because converting a Rust UUID back to the driver-compatible stdlib type is slower.

Parameters:

value (bytes) -- UUID bytes in big-endian order.

Return type:

UUID

Returns:

A standard-library UUID.

sqlspec.utils.uuids.uuid_from_int(value)[source]#

Construct a stdlib UUID from its 128-bit integer value.

Parameters:

value (int) -- UUID integer in the inclusive range 0 through 2**128 - 1.

Return type:

UUID

Returns:

A standard-library UUID.

sqlspec.utils.uuids.uuid_from_string(value)[source]#

Construct a stdlib UUID from text, using Rust parsing when available.

Parameters:

value (str) -- Canonical, hexadecimal, URN, or braced UUID text accepted by uuid.UUID and uuid_utils.UUID.

Return type:

UUID

Returns:

A standard-library UUID suitable for native database drivers.

The most common generators, uuid4, uuid6, uuid7, and nanoid, are also importable from the top-level sqlspec package.

Text and Identifiers#

Text processing utilities for SQLSpec.

Provides functions for string manipulation including case conversion, slugification, and email validation. Used primarily for identifier generation and data validation.

sqlspec.utils.text.camelize(string)[source]#

Convert a string to camel case.

Parameters:

string (str) -- The string to convert.

Return type:

str

Returns:

The converted string.

sqlspec.utils.text.kebabize(string)[source]#

Convert a string to kebab-case.

Parameters:

string (str) -- The string to convert.

Return type:

str

Returns:

The kebab-case version of the string.

sqlspec.utils.text.normalize_identifier(identifier, dialect)[source]#

Normalize an identifier-shaped value for dialect metadata lookups.

Parameters:

  • identifier (str) -- SQL identifier supplied by the caller.

  • dialect (str) -- SQL dialect name.

Return type:

str

Returns:

Identifier in the form expected by the dialect's metadata tables.

sqlspec.utils.text.pascalize(string)[source]#

Convert a string to PascalCase.

Parameters:

string (str) -- The string to convert.

Return type:

str

Returns:

The PascalCase version of the string.

sqlspec.utils.text.quote_backtick_identifier(identifier)[source]#

Quote a SQL identifier with backtick escaping (MySQL family).

Wraps the value in backticks and escapes any embedded backtick by doubling it, so each backtick becomes two. Used by MySQL-family adapters (asyncmy, aiomysql, mysqlconnector, pymysql) where the backtick is the dialect's identifier delimiter.

Parameters:

identifier (str) -- SQL identifier (schema, table, column, ...).

Return type:

str

Returns:

Backtick-quoted, escape-safe identifier.

sqlspec.utils.text.quote_identifier(identifier)[source]#

Quote a SQL identifier with double-quote escaping.

Wraps the value in double quotes and escapes any embedded double quote per the SQL standard (" -> ""). Used to safely interpolate identifier-shaped values (schemas, tables, columns) into SQL where bind parameters are not allowed (DDL identifiers, SET commands).

Dialect-aware case-folding is handled by normalize_identifier.

Parameters:

identifier (str) -- SQL identifier (schema, table, column, ...).

Return type:

str

Returns:

Double-quoted, escape-safe identifier.

sqlspec.utils.text.slugify(value, allow_unicode=False, separator=None)[source]#

Convert a string to a URL-friendly slug.

Parameters:

  • value (str) -- The string to slugify

  • allow_unicode (bool) -- Allow unicode characters in slug.

  • separator (str | None) -- Separator character for word boundaries. Defaults to "-".

Return type:

str

Returns:

A slugified string.

sqlspec.utils.text.snake_case(string)[source]#

Convert a string to snake_case.

Parameters:

string (str) -- The string to convert.

Return type:

str

Returns:

The snake_case version of the string.

sqlspec.utils.text.split_qualified_identifier(identifier, *, quote_chars='"`', allow_bracket_quotes=True)[source]#

Split a qualified SQL identifier on dots outside quoted parts.

Parameters:

  • identifier (str) -- SQL identifier or qualified identifier.

  • quote_chars (str) -- Quote characters to treat as identifier delimiters.

  • allow_bracket_quotes (bool) -- Treat SQL Server [...] as identifier quotes.

Return type:

tuple[str, ...]

Returns:

Identifier parts with outer identifier quotes removed and embedded quote escapes normalized.

Serialization#

Serialization utilities for SQLSpec.

class sqlspec.utils.serializers.SchemaSerializer[source]#

Bases: object

Serializer pipeline that caches conversions for repeated schema dumps.

__init__(key, dump)[source]#
sqlspec.utils.serializers.from_json(data, *, decode_bytes=True)#

Decode JSON input into Python data.

Return type:

Any

sqlspec.utils.serializers.get_collection_serializer(sample, *, exclude_unset=True, wire_format=False)[source]#

Return cached serializer pipeline for the provided sample object.

Return type:

SchemaSerializer

sqlspec.utils.serializers.get_serializer_metrics()[source]#

Return cache metrics aligned with the core pipeline counters.

Return type:

dict[str, int]

sqlspec.utils.serializers.numpy_array_dec_hook(target_or_value, value=<object object>)[source]#

Decode JSON list payloads into NumPy arrays.

Supports both direct one-argument usage and Litestar's (target_type, value) decoder contract.

Return type:

Any

sqlspec.utils.serializers.numpy_array_enc_hook(value)[source]#

Encode NumPy arrays and scalars to JSON-compatible values.

Return type:

Any

sqlspec.utils.serializers.reset_serializer_cache()[source]#

Clear cached serializer pipelines.

Return type:

None

sqlspec.utils.serializers.schema_dump(data, *, exclude_unset=True, wire_format=False)[source]#

Dump a schema model or dict to a plain representation.

Parameters:

  • data (Any) -- A schema instance (msgspec.Struct, Pydantic BaseModel, dataclass, attrs class) or plain dict / primitive.

  • exclude_unset (bool) -- If True, exclude fields that were never set (msgspec UNSET, Pydantic model_fields_set semantics). No-op for attrs (attrs has no unset concept).

  • wire_format (bool) -- msgspec-only knob. Default False emits Python attribute names (field.name) for cross-library consistency — keys match Pydantic, dataclass, and attrs output regardless of the Struct's rename= meta. Pass wire_format=True to emit field.encode_name (honours rename= on the Struct) for wire-aligned JSON / API payloads. Pydantic, dataclass, and attrs branches always use Python attribute names regardless of this flag.

Return type:

Any

sqlspec.utils.serializers.serialize_collection(items, *, exclude_unset=True, wire_format=False)[source]#

Serialize a collection using cached pipelines keyed by item type.

Return type:

list[typing.Any]

sqlspec.utils.serializers.to_json(data, *, as_bytes=False)#

Encode Python data into JSON.

Overloads:

  • data (Any), as_bytes (Literal[False]) → str

  • data (Any), as_bytes (Literal[True]) → bytes

sqlspec.utils.serializers.DEFAULT_TYPE_ENCODERS#

Default mapping of Python types to JSON encoder functions; type_encoders passed to a serializer are merged over it.

Schema Conversion#

Schema transformation utilities for converting data to various schema types.

sqlspec.utils.schema.to_schema(data, *, schema_type=None)[source]#

Convert data to a specified schema type.

Overloads:

  • data (list[DataT]), schema_type (type[SchemaT]) → list[SchemaT]

  • data (list[DataT]), schema_type (None) → list[DataT]

  • data (DataT), schema_type (type[SchemaT]) → SchemaT

  • data (DataT), schema_type (None) → DataT

Supports transformation to various schema types including:

  • TypedDict

  • dataclasses

  • msgspec Structs

  • Pydantic models

  • attrs classes

Parameters:

  • data (Any) -- Input data to convert (dict, list of dicts, or other)

  • schema_type (Any) -- Target schema type for conversion. If None, returns data unchanged.

Returns:

Converted data in the specified schema type, or original data if schema_type is None

Raises:

SQLSpecError -- If schema_type is not a supported type

sqlspec.utils.schema.to_value_type(value, value_type)[source]#

Convert a database value to the specified Python type.

This function handles type conversion for common database return values, providing runtime type safety for scalar queries. When the value is already the correct type, it is returned as-is without conversion overhead. Strict type identity check handles subclass gotchas (bool is subclass of int, datetime is subclass of date).

Also supports schema types (Pydantic models, dataclasses, msgspec Structs, attrs classes, and TypedDict). For schema types, JSON strings are automatically parsed before conversion.

Parameters:

  • value (Any) -- The value to convert.

  • value_type (type[TypeVar(ValueT)]) --

    The target Python type. Supported types include:

    • Primitives: int, float, str, bool

    • Temporal: datetime, date, time

    • Numeric: Decimal

    • Identifiers: UUID, Path

    • Collections: dict, list (for JSON/JSONB columns)

    • Schema types: Pydantic models, dataclasses, msgspec Structs, attrs classes, TypedDict (for JSONB columns)

Return type:

TypeVar(ValueT)

Returns:

The converted value of the specified type.

Raises:

TypeError -- If the value cannot be converted to the specified type.

sqlspec.utils.schema.transform_dict_keys(data, converter)[source]#

Transform dictionary keys using the provided converter function.

Recursively transforms all dictionary keys in a data structure using the provided converter function. Handles nested dictionaries, lists of dictionaries, and preserves non-dict values unchanged.

Parameters:

  • data (dict | list | Any) -- The data structure to transform. Can be a dict, list, or any other type.

  • converter (Callable[[str], str]) -- Function to convert string keys.

Return type:

dict | list | Any

Returns:

The transformed data structure with converted keys. Non-dict values are returned unchanged.

Correlation Tracking#

Correlation ID tracking for distributed tracing.

This module provides utilities for tracking correlation IDs across database operations, enabling distributed tracing and debugging.

class sqlspec.utils.correlation.CorrelationContext[source]#

Bases: object

Context manager for correlation ID tracking.

This class provides a context-aware way to track correlation IDs across async and sync operations.

classmethod get()[source]#

Get the current correlation ID.

Return type:

str | None

Returns:

The current correlation ID or None if not set

classmethod set(correlation_id)[source]#

Set the correlation ID.

Parameters:

correlation_id (str | None) -- The correlation ID to set

Return type:

None

classmethod generate()[source]#

Generate a new correlation ID.

Return type:

str

Returns:

A new UUID-based correlation ID

classmethod context(cls, correlation_id=None)[source]#

Context manager for correlation ID scope.

Parameters:

correlation_id (str | None) -- The correlation ID to use. If None, generates a new one.

Yields:

The correlation ID being used

classmethod clear()[source]#

Clear the current correlation ID.

Return type:

None

classmethod to_dict()[source]#

Get correlation context as a dictionary.

Return type:

dict[str, typing.Any]

Returns:

Dictionary with correlation_id key if set

sqlspec.utils.correlation.correlation_context(correlation_id=None)[source]#

Convenience context manager for correlation ID tracking.

Parameters:

correlation_id (str | None) -- Optional correlation ID. If None, generates a new one.

Yields:

The active correlation ID

Example

with correlation_context() as correlation_id:
    logger.info(
        "Processing request",
        extra={"correlation_id": correlation_id},
    )
sqlspec.utils.correlation.get_correlation_adapter(logger)[source]#

Get a logger adapter that automatically includes correlation ID.

Return type:

LoggerAdapter

Module and Optional Dependency Loading#

Module loading utilities for SQLSpec.

Provides functions for dynamic module imports, path resolution, and dependency availability checking. Used for loading modules from dotted paths, converting module paths to filesystem paths, and ensuring optional dependencies are installed.

class sqlspec.utils.module_loader.OptionalDependencyFlag[source]#

Bases: object

Boolean-like wrapper that evaluates module availability lazily.

__init__(module_name)[source]#
sqlspec.utils.module_loader.dependency_flag(module_name)[source]#

Return a lazily evaluated flag for the supplied module name.

Parameters:

module_name (str) -- Dotted module path to guard.

Return type:

OptionalDependencyFlag

Returns:

OptionalDependencyFlag tracking the module.

sqlspec.utils.module_loader.ensure_attrs()[source]#

Ensure attrs is available.

Return type:

None

sqlspec.utils.module_loader.ensure_cattrs()[source]#

Ensure cattrs is available.

Return type:

None

sqlspec.utils.module_loader.ensure_fsspec()[source]#

Ensure fsspec is available for filesystem operations.

Return type:

None

sqlspec.utils.module_loader.ensure_litestar()[source]#

Ensure Litestar is available.

Return type:

None

sqlspec.utils.module_loader.ensure_msgspec()[source]#

Ensure msgspec is available for serialization.

Return type:

None

sqlspec.utils.module_loader.ensure_numpy()[source]#

Ensure NumPy is available for array operations.

Return type:

None

sqlspec.utils.module_loader.ensure_obstore()[source]#

Ensure obstore is available for object storage operations.

Return type:

None

sqlspec.utils.module_loader.ensure_opentelemetry()[source]#

Ensure OpenTelemetry is available for tracing.

Return type:

None

sqlspec.utils.module_loader.ensure_orjson()[source]#

Ensure orjson is available for fast JSON operations.

Return type:

None

sqlspec.utils.module_loader.ensure_pandas()[source]#

Ensure pandas is available for DataFrame operations.

Return type:

None

sqlspec.utils.module_loader.ensure_pgvector()[source]#

Ensure pgvector is available for vector operations.

Return type:

None

sqlspec.utils.module_loader.ensure_polars()[source]#

Ensure Polars is available for DataFrame operations.

Return type:

None

sqlspec.utils.module_loader.ensure_prometheus()[source]#

Ensure Prometheus client is available for metrics.

Return type:

None

sqlspec.utils.module_loader.ensure_pyarrow()[source]#

Ensure PyArrow is available for Arrow operations.

Return type:

None

sqlspec.utils.module_loader.ensure_pydantic()[source]#

Ensure Pydantic is available for data validation.

Return type:

None

sqlspec.utils.module_loader.ensure_uvloop()[source]#

Ensure uvloop is available for fast event loops.

Return type:

None

sqlspec.utils.module_loader.import_optional(module_name)[source]#

Return the imported module if available, otherwise None.

Silent-fallback twin of the ensure_pyarrow()-style helpers: use it when the caller has a working fallback (e.g. stdlib uuid) and must not raise on a missing optional dependency. The resolved module (or None) is cached per interpreter session; call reset_dependency_cache() to invalidate it.

Parameters:

module_name (str) -- Dotted module path to import.

Return type:

ModuleType | None

Returns:

The imported module, or None when it is not installed.

sqlspec.utils.module_loader.import_optional_attr(module_name, attr)[source]#

Return getattr(module, attr) when the module imports, else None.

Companion to import_optional() for acquiring a submodule, class, or constant from an optional dependency (e.g. pgvector.asyncpg) without raising when the dependency or attribute is absent.

Parameters:

  • module_name (str) -- Dotted module path to import.

  • attr (str) -- Attribute name to resolve on the imported module.

Return type:

Any

Returns:

The resolved attribute, or None when the module or attribute is unavailable.

sqlspec.utils.module_loader.import_string(dotted_path)[source]#

Import a module or attribute from a dotted path string.

Failures are reported as ImportError, except for SQLSpec errors raised by the imported module itself, which propagate unchanged so their own message survives. MissingDependencyError is reported as ImportError because it describes an import failure.

Parameters:

dotted_path (str) -- The path of the module to import.

Return type:

typing.Any

Returns:

The imported object.

sqlspec.utils.module_loader.module_available(module_name)[source]#

Return True if the given module can be resolved.

The result is cached per interpreter session. Call reset_dependency_cache() to invalidate cached entries when tests manipulate sys.path.

Parameters:

module_name (str) -- Dotted module path to check.

Return type:

bool

Returns:

True if importlib can find the module, False otherwise.

sqlspec.utils.module_loader.module_to_os_path(dotted_path='app')[source]#

Convert a module dotted path to filesystem path.

Namespace packages have no origin; their first search location is used instead.

Parameters:

dotted_path (str) -- The path to the module.

Raises:

ModuleNotFoundError -- The module could not be found.

Return type:

Path

Returns:

The path to the module.

sqlspec.utils.module_loader.reset_dependency_cache(module_name=None)[source]#

Clear cached availability for one module or the entire cache.

Parameters:

module_name (str | None) -- Specific dotted module path to drop from the cache. Clears the full cache when None.

Return type:

None

sqlspec.utils.module_loader.resolve_optional_attr(module_name, attr, fallback)[source]#

Return an optional module or attribute, otherwise the provided fallback.

Unlike import_optional_attr(), this keeps the caller-supplied shim object stable so module-level caches can preserve identity when the optional dependency is missing.

Parameters:

  • module_name (str) -- Dotted module path to import.

  • attr (str | None) -- Attribute name to resolve on the imported module. Returns the module itself when None.

  • fallback (TypeVar(T)) -- Shim object to return when the module or attribute is missing.

Return type:

TypeVar(T)

Returns:

The resolved attribute or the fallback shim.

Sync and Async Interoperation#

The executor controls are process-wide knobs. Use them when adapting a synchronous driver or callback to an async application; ordinary application code should prefer its framework's native concurrency primitives.

Utilities for async/sync interoperability in SQLSpec.

This module provides utilities for converting between async and sync functions, managing concurrency limits, and handling context managers. Used primarily for adapter implementations that need to support both sync and async patterns.

class sqlspec.utils.sync_tools.NoValue[source]#

Bases: object

Sentinel class for missing values.

class sqlspec.utils.sync_tools.CapacityLimiter[source]#

Bases: object

Limits the number of concurrent operations using a semaphore.

__init__(total_tokens)[source]#

Initialize the capacity limiter.

Parameters:

total_tokens (int) -- Maximum number of concurrent operations allowed

async acquire()[source]#

Acquire a token from the semaphore.

Return type:

None

release()[source]#

Release a token back to the semaphore.

Return type:

None

property total_tokens: int#

Get the total number of tokens available.

async __aenter__()[source]#

Async context manager entry.

Return type:

None

async __aexit__(exc_type, exc_val, exc_tb)[source]#

Async context manager exit.

Return type:

None

sqlspec.utils.sync_tools.set_default_async_executor(executor)[source]#

Set a caller-owned process-local default thread executor for async_.

SQLSpec never shuts down caller-provided executors. If a process fork is detected later, the stored executor is cleared instead of reused, and async_ falls back to SQLSpec's managed default pool.

Parameters:

executor (ThreadPoolExecutor | None) -- Caller-owned thread executor to use by default, or None to clear it and use SQLSpec's managed pool.

Return type:

None

sqlspec.utils.sync_tools.enable_default_async_thread_pool(max_workers=None)[source]#

Configure and eagerly create SQLSpec's managed default executor for async_.

Parameters:

max_workers (int | None) -- Optional worker limit. When omitted, SQLSPEC_ASYNC_THREAD_LIMIT is read, falling back to DEFAULT_ASYNC_THREAD_LIMIT.

Return type:

None

sqlspec.utils.sync_tools.get_default_async_executor()[source]#

Return the configured or managed default executor for async_.

Return type:

ThreadPoolExecutor

Returns:

Caller-owned default thread executor or SQLSpec-managed executor.

sqlspec.utils.sync_tools.shutdown_default_async_executor(wait=False)[source]#

Clear default async executors and shut down SQLSpec-owned pools.

The next async_ call recreates SQLSpec's managed default pool.

Parameters:

wait (bool) -- Whether to wait for SQLSpec-managed executor tasks to finish.

Return type:

None

sqlspec.utils.sync_tools.run_(async_function)[source]#

Convert an async function to a blocking function using asyncio.run().

Parameters:

async_function (typing.Callable[ParamSpec(ParamSpecT, bound= None), Coroutine[typing.Any, typing.Any, TypeVar(ReturnT)]]) -- The async function to convert.

Return type:

typing.Callable[ParamSpec(ParamSpecT, bound= None), TypeVar(ReturnT)]

Returns:

A blocking function that runs the async function.

sqlspec.utils.sync_tools.await_(async_function, raise_sync_error=False)[source]#

Convert an async function to a blocking one, running in the main async loop.

When no event loop exists, automatically creates and uses a global portal for async-to-sync bridging via background thread. Set raise_sync_error=True to disable this behavior and raise errors instead.

Parameters:

  • async_function (typing.Callable[ParamSpec(ParamSpecT, bound= None), Coroutine[typing.Any, typing.Any, TypeVar(ReturnT)]]) -- The async function to convert.

  • raise_sync_error (bool) -- If True, raises RuntimeError when no loop exists. If False (default), uses portal pattern for automatic bridging.

Return type:

typing.Callable[ParamSpec(ParamSpecT, bound= None), TypeVar(ReturnT)]

Returns:

A blocking function that runs the async function.

sqlspec.utils.sync_tools.async_(function, *, limiter=None, executor=None)[source]#

Convert a blocking function to an async one using a thread executor.

Parameters:
Return type:

typing.Callable[ParamSpec(ParamSpecT, bound= None), Awaitable[TypeVar(ReturnT)]]

Returns:

An async function that runs the original function in a thread.

sqlspec.utils.sync_tools.ensure_async_(function)[source]#

Convert a function to an async one if it is not already.

Overloads:

  • function (Callable[ParamSpecT, Awaitable[ReturnT]]) → Callable[ParamSpecT, Coroutine[Any, Any, ReturnT]]

  • function (Callable[ParamSpecT, Awaitable[ReturnT] | ReturnT]) → Callable[ParamSpecT, Coroutine[Any, Any, ReturnT]]

Parameters:

function (typing.Callable[ParamSpec(ParamSpecT, bound= None), typing.Any]) -- The function to convert.

Returns:

An async function that runs the original function.

sqlspec.utils.sync_tools.with_ensure_async_(obj)[source]#

Convert a context manager to an async one if it is not already.

Parameters:

obj (AbstractContextManager[TypeVar(T)] | AbstractAsyncContextManager[TypeVar(T)]) -- The context manager to convert.

Return type:

AbstractAsyncContextManager[TypeVar(T)]

Returns:

An async context manager that runs the original context manager.

async sqlspec.utils.sync_tools.get_next(iterable, default=<sqlspec.utils.sync_tools.NoValue object>, *args)[source]#

Return the next item from an async iterator.

Parameters:

  • iterable (Any) -- An async iterable.

  • default (Any) -- An optional default value to return if the iterable is empty.

  • *args (Any) -- The remaining args

Return type:

Any

Returns:

The next value of the iterable.

Deprecation Helpers#

Deprecation utilities for SQLSpec.

Provides decorators and warning functions for marking deprecated functionality. Used to communicate API changes and migration paths to users.

sqlspec.utils.deprecation.deprecated(version, *, removal_in=None, alternative=None, info=None, pending=False, kind=None)[source]#

Create a decorator wrapping a function, method or property with a deprecation warning.

Parameters:

  • version (str) -- SQLSpec version where the deprecation will occur

  • removal_in (str | None) -- SQLSpec version where the deprecated function will be removed

  • alternative (str | None) -- Name of a function that should be used instead

  • info (str | None) -- Additional information

  • pending (bool) -- Use warnings.PendingDeprecationWarning instead of warnings.DeprecationWarning

  • kind (Optional[Literal['function', 'method', 'classmethod', 'property']]) -- Type of the deprecated callable. If None, will use inspect to figure out if it's a function or method

Return type:

Callable[[Callable[[ParamSpec(P, bound= None)], TypeVar(T)]], Callable[[ParamSpec(P, bound= None)], TypeVar(T)]]

Returns:

A decorator wrapping the function call with a warning

sqlspec.utils.deprecation.warn_deprecation(version, deprecated_name, kind, *, removal_in=None, alternative=None, info=None, pending=False)[source]#

Warn about a call to a deprecated function, attributed to the first caller outside SQLSpec.

Parameters:

  • version (str) -- SQLSpec version where the deprecation will occur

  • deprecated_name (str) -- Name of the deprecated function

  • removal_in (str | None) -- SQLSpec version where the deprecated function will be removed

  • alternative (str | None) -- Name of a function that should be used instead

  • info (str | None) -- Additional information

  • pending (bool) -- Use warnings.PendingDeprecationWarning instead of warnings.DeprecationWarning

  • kind (Literal['function', 'method', 'classmethod', 'attribute', 'property', 'class', 'parameter', 'import']) -- Type of the deprecated thing

Return type:

None

Fixture Files#

Fixture loading utilities for SQLSpec.

Provides functions for writing, loading and parsing JSON fixture files used in testing and development, and for loading and exporting per-table fixture files against a database driver. Supports both sync and async operations.

async sqlspec.utils.fixtures.export_table_fixtures_async(driver, fixtures_path, tables, *, compress=True, jsonl=False)[source]#

Export database tables to per-table fixture files asynchronously.

Writes every row of each table to <table>.json (a JSON array) or, with jsonl, <table>.jsonl (one JSON object per line), gzipped with a .gz suffix when compress is set. Generated columns are left out. Rows are ordered by the primary key, or by every orderable column when the table has none, and each table is read into memory. Dates, times, and datetimes are written as ISO 8601 strings, Decimal and UUID values as strings, and bytes as base64 strings, which load_table_fixtures_async() converts back for typed columns.

Each file is written to a temporary file in the directory, created with the process umask, and then moved into place; after that, other table fixture files for the same table in the directory are removed, so the next load reads the exported file. The directory is created when missing, and file writes run in a worker thread. Table names are quoted and must match the database spelling exactly.

Parameters:

  • driver (AsyncDriverAdapterBase) -- Async driver that reads the tables.

  • fixtures_path (str | Path) -- Directory to write the fixture files to.

  • tables (Sequence[str]) -- Tables to export.

  • compress (bool) -- Whether to gzip the written files.

  • jsonl (bool) -- Whether to write one JSON object per line instead of a JSON array.

Return type:

dict[str, int]

Returns:

Mapping of table name to the number of rows written.

Raises:

ValueError -- If a table name is not a plain SQL identifier.

sqlspec.utils.fixtures.export_table_fixtures_sync(driver, fixtures_path, tables, *, compress=True, jsonl=False)[source]#

Export database tables to per-table fixture files.

Writes every row of each table to <table>.json (a JSON array) or, with jsonl, <table>.jsonl (one JSON object per line), gzipped with a .gz suffix when compress is set. Generated columns are left out. Rows are ordered by the primary key, or by every orderable column when the table has none, and each table is read into memory. Dates, times, and datetimes are written as ISO 8601 strings, Decimal and UUID values as strings, and bytes as base64 strings, which load_table_fixtures_sync() converts back for typed columns.

Each file is written to a temporary file in the directory, created with the process umask, and then moved into place; after that, other table fixture files for the same table in the directory are removed, so the next load reads the exported file. The directory is created when missing. Table names are quoted and must match the database spelling exactly.

Parameters:

  • driver (SyncDriverAdapterBase) -- Sync driver that reads the tables.

  • fixtures_path (str | Path) -- Directory to write the fixture files to.

  • tables (Sequence[str]) -- Tables to export.

  • compress (bool) -- Whether to gzip the written files.

  • jsonl (bool) -- Whether to write one JSON object per line instead of a JSON array.

Return type:

dict[str, int]

Returns:

Mapping of table name to the number of rows written.

Raises:

ValueError -- If a table name is not a plain SQL identifier.

async sqlspec.utils.fixtures.load_table_fixtures_async(driver, fixtures_path, *, tables=None, table_order=None, conflict_keys=None, batch_size=500, resync_sequences=False, ignore_unknown_columns=False, exclude_update_columns=None)[source]#

Load per-table fixture files into database tables asynchronously.

Each table reads from one file named after it: <table>.json holding a JSON array of row objects, or <table>.jsonl holding one row object per line, either optionally gzipped (.gz). Sparse row objects are normalized across the ordered union of column keys in the file, defaulting omitted keys to None. Table and column names are quoted, so they must match the database spelling exactly. Rows are inserted in batches with execute_many; each file is read into memory in a worker thread. Transaction control stays with the caller.

On PostgreSQL-family, MySQL, DuckDB, and SQLite drivers the target table's columns are read from the driver's data dictionary first, and JSON values of these column types are converted: ISO 8601 strings in timestamp, datetime, date, and (except on MySQL) time columns to datetimes, dates, and times; ISO 8601 durations without year or month parts and numbers of seconds in interval columns to timedelta, and DuckDB [months, days, nanoseconds] interval lists to interval text; strings and numbers in numeric and decimal columns to Decimal; strings in uuid columns to UUID; base64 strings in bytea, blob, binary, and varbinary columns to bytes; and PostgreSQL, MySQL, DuckDB, and SQLite json and jsonb values to JSON text. SQLite only converts binary and JSON columns. All other values, including array elements and durations with year or month parts such as P1M, are passed to the driver as decoded from JSON. Values of generated columns are ignored. On PostgreSQL, GENERATED ALWAYS identity columns receive the loaded values through OVERRIDING SYSTEM VALUE and are never updated by upserts.

Parameters:

  • driver (AsyncDriverAdapterBase) -- Async driver that runs the inserts.

  • fixtures_path (str | Path) -- Directory containing the table fixture files.

  • tables (Sequence[str] | None) -- Tables to load. Defaults to every table fixture file in the directory; files whose names are not table identifiers are skipped with a debug log, and a dot in a file name marks a schema-qualified table.

  • table_order (Sequence[str] | None) -- Tables to load first, in this order. Remaining tables load afterwards in alphabetical order; names not being loaded are ignored.

  • conflict_keys (Mapping[str, Sequence[str] | str] | None) -- Mapping of table name to a column name or sequence of key column names of a unique constraint; tables not being loaded are ignored. Rows for matched tables are upserted, updating every non-key, non-excluded column when a row with the same key already exists. Supported on PostgreSQL-family, SQLite, and DuckDB drivers (ON CONFLICT) and MySQL (ON DUPLICATE KEY UPDATE, which matches any unique key of the table).

  • batch_size (int) -- Maximum number of rows per execute_many call.

  • resync_sequences (bool) -- On PostgreSQL-family drivers, set the sequence behind each serial or identity column of a loaded table to the column's maximum value, or back to the sequence's minimum value for an empty table, so the next generated value follows the loaded rows. On other dialects the option is skipped with a debug log.

  • ignore_unknown_columns (bool) -- When True and table column metadata is available, strip fixture columns not present in the target table before inserting.

  • exclude_update_columns (Sequence[str] | Mapping[str, Sequence[str] | str] | None) -- Column names (or per-table mapping of column names) to omit from the SET clause of upserts so existing values are preserved on conflict.

Return type:

dict[str, int]

Returns:

Mapping of table name to the number of rows loaded, in load order.

Raises:

  • ValueError -- If batch_size is below 1, conflict_keys is given for an unsupported dialect, a table or column name is not a plain SQL identifier, a table has more than one fixture file, a row in a fixture file is empty, or a fixture file only holds generated columns.

  • TypeError -- If a fixture file does not hold a list of row objects, or conflict_keys / exclude_update_columns has an invalid type.

  • FileNotFoundError -- If a requested table has no fixture file.

sqlspec.utils.fixtures.load_table_fixtures_sync(driver, fixtures_path, *, tables=None, table_order=None, conflict_keys=None, batch_size=500, resync_sequences=False, ignore_unknown_columns=False, exclude_update_columns=None)[source]#

Load per-table fixture files into database tables.

Each table reads from one file named after it: <table>.json holding a JSON array of row objects, or <table>.jsonl holding one row object per line, either optionally gzipped (.gz). Sparse row objects are normalized across the ordered union of column keys in the file, defaulting omitted keys to None. Table and column names are quoted, so they must match the database spelling exactly. Rows are inserted in batches with execute_many; each file is read into memory. Transaction control stays with the caller.

On PostgreSQL-family, MySQL, DuckDB, and SQLite drivers the target table's columns are read from the driver's data dictionary first, and JSON values of these column types are converted: ISO 8601 strings in timestamp, datetime, date, and (except on MySQL) time columns to datetimes, dates, and times; ISO 8601 durations without year or month parts and numbers of seconds in interval columns to timedelta, and DuckDB [months, days, nanoseconds] interval lists to interval text; strings and numbers in numeric and decimal columns to Decimal; strings in uuid columns to UUID; base64 strings in bytea, blob, binary, and varbinary columns to bytes; and PostgreSQL, MySQL, DuckDB, and SQLite json and jsonb values to JSON text. SQLite only converts binary and JSON columns. All other values, including array elements and durations with year or month parts such as P1M, are passed to the driver as decoded from JSON. Values of generated columns are ignored. On PostgreSQL, GENERATED ALWAYS identity columns receive the loaded values through OVERRIDING SYSTEM VALUE and are never updated by upserts.

Parameters:

  • driver (SyncDriverAdapterBase) -- Sync driver that runs the inserts.

  • fixtures_path (str | Path) -- Directory containing the table fixture files.

  • tables (Sequence[str] | None) -- Tables to load. Defaults to every table fixture file in the directory; files whose names are not table identifiers are skipped with a debug log, and a dot in a file name marks a schema-qualified table.

  • table_order (Sequence[str] | None) -- Tables to load first, in this order. Remaining tables load afterwards in alphabetical order; names not being loaded are ignored.

  • conflict_keys (Mapping[str, Sequence[str] | str] | None) -- Mapping of table name to a column name or sequence of key column names of a unique constraint; tables not being loaded are ignored. Rows for matched tables are upserted, updating every non-key, non-excluded column when a row with the same key already exists. Supported on PostgreSQL-family, SQLite, and DuckDB drivers (ON CONFLICT) and MySQL (ON DUPLICATE KEY UPDATE, which matches any unique key of the table).

  • batch_size (int) -- Maximum number of rows per execute_many call.

  • resync_sequences (bool) -- On PostgreSQL-family drivers, set the sequence behind each serial or identity column of a loaded table to the column's maximum value, or back to the sequence's minimum value for an empty table, so the next generated value follows the loaded rows. On other dialects the option is skipped with a debug log.

  • ignore_unknown_columns (bool) -- When True and table column metadata is available, strip fixture columns not present in the target table before inserting.

  • exclude_update_columns (Sequence[str] | Mapping[str, Sequence[str] | str] | None) -- Column names (or per-table mapping of column names) to omit from the SET clause of upserts so existing values are preserved on conflict.

Return type:

dict[str, int]

Returns:

Mapping of table name to the number of rows loaded, in load order.

Raises:

  • ValueError -- If batch_size is below 1, conflict_keys is given for an unsupported dialect, a table or column name is not a plain SQL identifier, a table has more than one fixture file, a row in a fixture file is empty, or a fixture file only holds generated columns.

  • TypeError -- If a fixture file does not hold a list of row objects, or conflict_keys / exclude_update_columns has an invalid type.

  • FileNotFoundError -- If a requested table has no fixture file.

async sqlspec.utils.fixtures.open_fixture_async(fixtures_path, fixture_name)[source]#

Load and parse a JSON fixture file asynchronously with compression support.

Supports reading from:

  • Regular JSON files (.json)

  • Gzipped JSON files (.json.gz)

  • Zipped JSON files (.json.zip)

For compressed files, uses sync reading in a thread pool since gzip and zipfile don't have native async equivalents.

Parameters:

  • fixtures_path (Any) -- The path to look for fixtures (pathlib.Path)

  • fixture_name (str) -- The fixture name to load.

Return type:

Any

Returns:

The parsed JSON data

sqlspec.utils.fixtures.open_fixture_sync(fixtures_path, fixture_name)[source]#

Load and parse a JSON fixture file with compression support.

Supports reading from:

  • Regular JSON files (.json)

  • Gzipped JSON files (.json.gz)

  • Zipped JSON files (.json.zip)

Parameters:

  • fixtures_path (Any) -- The path to look for fixtures (pathlib.Path)

  • fixture_name (str) -- The fixture name to load.

Return type:

Any

Returns:

The parsed JSON data

async sqlspec.utils.fixtures.write_fixture_async(fixtures_path, table_name, data, storage_backend='local', compress=False, **storage_kwargs)[source]#

Write fixture data to storage using SQLSpec storage backend asynchronously.

Parameters:

  • fixtures_path (str) -- Base path where fixtures should be stored

  • table_name (str) -- Name of the table/fixture (used as filename)

  • data (list[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]] | list[dict[str, typing.Any]] | DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]) -- Data to write - can be list of dicts, models, or single model

  • storage_backend (str) -- Storage backend to use (default: "local")

  • compress (bool) -- Whether to gzip compress the output

  • **storage_kwargs (Any) -- Additional arguments for the storage backend

Raises:

ValueError -- If storage backend is not found

Return type:

None

sqlspec.utils.fixtures.write_fixture_sync(fixtures_path, table_name, data, storage_backend='local', compress=False, **storage_kwargs)[source]#

Write fixture data to storage using SQLSpec storage backend.

Parameters:

  • fixtures_path (str) -- Base path where fixtures should be stored

  • table_name (str) -- Name of the table/fixture (used as filename)

  • data (list[DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]] | list[dict[str, typing.Any]] | DictLike | StructStub | BaseModelStub | DataclassProtocol | AttrsInstanceStub | Mapping[str, Any]) -- Data to write - can be list of dicts, models, or single model

  • storage_backend (str) -- Storage backend to use (default: "local")

  • compress (bool) -- Whether to gzip compress the output

  • **storage_kwargs (Any) -- Additional arguments for the storage backend

Raises:

ValueError -- If storage backend is not found

Return type:

None