Utilities¶
from sqlspec.utils.arrow_helpers import (
coerce_arrow_batches,
coerce_arrow_table,
encode_arrow_batch_pg_binary,
iter_pgcopy_binary_chunks,
)
from sqlspec.utils.correlation import CorrelationContext
from sqlspec.utils.deprecation import deprecated
from sqlspec.utils.env import get_env
from sqlspec.utils.fixtures import (
export_table_fixtures_async,
export_table_fixtures_sync,
load_table_fixtures_async,
load_table_fixtures_sync,
)
from sqlspec.utils.module_loader import import_optional, import_string
from sqlspec.utils.schema import to_schema
from sqlspec.utils.serializers import from_json, to_json
from sqlspec.utils.sync_tools import async_, run_
from sqlspec.utils.text import camelize, snake_case
from sqlspec.utils.type_guards import (
is_msgspec_struct,
is_pydantic_model,
is_schema,
)
from sqlspec.utils.uuids import nanoid, uuid4, uuid6, uuid7
Reusable helpers for runtime type narrowing, JSON serialization, schema model conversion, UUID/NanoID generation, Apache Arrow batch/table coercion, table fixture loading/exporting, correlation tracking, and sync/async bridging.
Schema, Serialization & Identifiers¶
to_schema(data, *, schema_type=None)
¶
Convert data to a specified schema type.
Supports transformation to various schema types including
- TypedDict
- dataclasses
- msgspec Structs
- Pydantic models
- attrs classes
Returns:
| Type | Description |
|---|---|
Any
|
Converted data in the specified schema type, or original data if schema_type is None |
Raises:
| Type | Description |
|---|---|
SQLSpecError
|
If schema_type is not a supported type |
to_json(data, *, as_bytes=False)
¶
Encode Python data into JSON.
from_json(data, *, decode_bytes=True)
¶
Decode JSON input into Python data.
uuid4()
¶
Generate a random UUID (version 4).
Uses uuid-utils for performance when available, falls back to standard library uuid.uuid4() silently (equivalent output).
Returns:
| Type | Description |
|---|---|
UUID
|
A randomly generated UUID. |
uuid6()
¶
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.
Returns:
| Type | Description |
|---|---|
UUID
|
A time-ordered UUID, or a random UUID if time-ordered generation unavailable. |
uuid7()
¶
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.
Returns:
| Type | Description |
|---|---|
UUID
|
A time-ordered UUID, or a random UUID if time-ordered generation unavailable. |
nanoid()
¶
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.
Returns:
| Type | Description |
|---|---|
str
|
A 21-character Nano ID string, or 32-character UUID hex if |
str
|
fastnanoid unavailable. |
camelize(string)
¶
snake_case(string)
¶
Arrow Coercion & Binary Helpers¶
coerce_arrow_table(source)
¶
Coerce various sources to a PyArrow Table.
coerce_arrow_batches(source, *, batch_size=65536)
¶
Coerce an Arrow source into column names and a lazy RecordBatch iterator.
Streams pyarrow.RecordBatchReader and __arrow_c_stream__ sources one
batch at a time without materializing the full table into memory.
Returns:
| Type | Description |
|---|---|
tuple[list[str], Iterator[ArrowRecordBatch]]
|
Tuple of |
encode_arrow_batch_pg_binary(batch, *, columns=None, vector_columns=frozenset(), jsonb_columns=frozenset(), include_header=False, include_trailer=False)
¶
Encode an Arrow RecordBatch into PostgreSQL COPY BINARY format.
Uses zero-loop structured NumPy dtype packing when all columns have fixed wire widths and no nulls, falling back to direct Arrow buffer slicing for variable-width and nullable columns.
Returns:
| Type | Description |
|---|---|
bytes
|
PostgreSQL |
iter_pgcopy_binary_chunks(source, *, columns=None, batch_size=65536, chunk_bytes=8388608, vector_columns=frozenset(), jsonb_columns=frozenset(), metrics=None)
¶
Yield PostgreSQL COPY BINARY chunks from an Arrow source in bounded memory.
Yields:
| Type | Description |
|---|---|
bytes
|
Byte chunks forming a complete |
Table Fixtures & Environment Helpers¶
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)
¶
Load per-table fixture files into database tables.
Each table reads from one or more files 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). Tables split into multiple shards (e.g. <table>.001.jsonl.gz,
<table>.002.jsonl.gz) are loaded in ascending numeric shard order. 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.
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 (datetimes on Oracle, whose DATE carries a time), 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, raw, image, binary, and varbinary columns
to bytes; and PostgreSQL, MySQL, DuckDB, and SQLite json and jsonb values to JSON
text. Vector column values are converted according to the driver family and data
type. Values of generated (computed) columns are ignored, while identity column
values are loaded. On PostgreSQL, GENERATED ALWAYS identity columns receive
the loaded values through OVERRIDING SYSTEM VALUE and are never updated by upserts;
on SQL Server, inserts into identity columns run between SET IDENTITY_INSERT ... ON
and OFF. Other databases receive identity values as given and reject them for
identity columns that only allow generated values.
Returns:
| Type | Description |
|---|---|
dict[str, int]
|
Mapping of table name to the number of rows loaded, in load order. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
TypeError
|
If a fixture file does not hold a list of row objects, or
|
FileNotFoundError
|
If a requested table has no fixture file. |
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)
¶
Load per-table fixture files into database tables asynchronously.
Each table reads from one or more files 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). Tables split into multiple shards (e.g. <table>.001.jsonl.gz,
<table>.002.jsonl.gz) are loaded in ascending numeric shard order. 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.
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 (datetimes on Oracle, whose DATE carries a time), 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, raw, image, binary, and varbinary columns
to bytes; and PostgreSQL, MySQL, DuckDB, and SQLite json and jsonb values to JSON
text. Vector column values are converted according to the driver family and data
type. Values of generated (computed) columns are ignored, while identity column
values are loaded. On PostgreSQL, GENERATED ALWAYS identity columns receive
the loaded values through OVERRIDING SYSTEM VALUE and are never updated by upserts;
on SQL Server, inserts into identity columns run between SET IDENTITY_INSERT ... ON
and OFF. Other databases receive identity values as given and reject them for
identity columns that only allow generated values.
Returns:
| Type | Description |
|---|---|
dict[str, int]
|
Mapping of table name to the number of rows loaded, in load order. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
TypeError
|
If a fixture file does not hold a list of row objects, or
|
FileNotFoundError
|
If a requested table has no fixture file. |
export_table_fixtures_sync(driver, fixtures_path, tables, *, compress=True, jsonl=False, max_shard_bytes=None, max_rows_per_shard=None)
¶
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. When max_shard_bytes or max_rows_per_shard
is provided and the table exceeds the threshold, rows are split across numbered
shards <table>.001<ext>, <table>.002<ext>, etc. 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, bytes as base64 strings, and vector
values as lists of numbers, which :func: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 stale table fixture files for the same table in the directory are removed before moving the new files into place, so the next load reads the exported files. The directory is created when missing. Table names are quoted and must match the database spelling exactly.
Returns:
| Type | Description |
|---|---|
dict[str, int]
|
Mapping of table name to the number of rows written. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a table name is not a plain SQL identifier, or shard limits are invalid. |
export_table_fixtures_async(driver, fixtures_path, tables, *, compress=True, jsonl=False, max_shard_bytes=None, max_rows_per_shard=None)
¶
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. When max_shard_bytes or max_rows_per_shard
is provided and the table exceeds the threshold, rows are split across numbered
shards <table>.001<ext>, <table>.002<ext>, etc. 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, bytes as base64 strings, and vector
values as lists of numbers, which :func: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 stale table fixture files for the same table in the directory are removed before moving the new files into place, so the next load reads the exported files. 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.
Returns:
| Type | Description |
|---|---|
dict[str, int]
|
Mapping of table name to the number of rows written. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If a table name is not a plain SQL identifier, or shard limits are invalid. |
get_env(key, default, type_hint=_UNSET)
¶
Return a callable that parses an environment variable on demand.
Returns:
| Type | Description |
|---|---|
Callable[[], Any]
|
Callable that returns the parsed value. |
Type Guards & Module Loading¶
is_msgspec_struct(obj)
¶
Check if a value is a msgspec struct class or instance.
Returns:
| Type | Description |
|---|---|
TypeGuard[StructStub]
|
bool |
is_pydantic_model(obj)
¶
Check if a value is a pydantic model class or instance.
Returns:
| Type | Description |
|---|---|
TypeGuard[BaseModelStub]
|
bool |
is_schema(obj)
¶
Check if a value is a msgspec Struct, Pydantic model, attrs instance, or schema class.
Returns:
| Type | Description |
|---|---|
TypeGuard[SupportedSchemaModel]
|
bool |
import_string(dotted_path)
¶
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.
Returns:
| Type | Description |
|---|---|
Any
|
The imported object. |
import_optional(module_name)
¶
Return the imported module if available, otherwise None.
Silent-fallback twin of the :func: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 :func:reset_dependency_cache to
invalidate it.
Returns:
| Type | Description |
|---|---|
ModuleType | None
|
The imported module, or |
Sync/Async Bridging, Correlation & Deprecation¶
run_(async_function)
¶
Convert an async function to a blocking function using asyncio.run().
Returns:
| Type | Description |
|---|---|
Callable[ParamSpecT, ReturnT]
|
A blocking function that runs the async function. |
async_(function, *, limiter=None, executor=None)
¶
Convert a blocking function to an async one using a thread executor.
Returns:
| Type | Description |
|---|---|
Callable[ParamSpecT, Awaitable[ReturnT]]
|
An async function that runs the original function in a thread. |
Context manager for correlation ID tracking.
This class provides a context-aware way to track correlation IDs across async and sync operations.
deprecated(version, *, removal_in=None, alternative=None, info=None, pending=False, kind=None)
¶
Create a decorator wrapping a function, method or property with a deprecation warning.
Returns:
| Type | Description |
|---|---|
Callable[[Callable[P, T]], Callable[P, T]]
|
A decorator wrapping the function call with a warning |