DDL¶

Data Definition Language builders for tables, indexes, views, and schemas.

Base¶

class sqlspec.builder.DDLBuilder[source]¶

Bases: QueryBuilder

Base class for DDL builders (CREATE, DROP, ALTER, etc).

__init__(dialect=None)[source]¶

build(dialect=None)[source]¶

Builds the SQL query string and parameters.

Parameters:: dialect¶ (Union[str, Dialect, Type[Dialect], None]) – Optional dialect override. If provided, generates SQL for this dialect instead of the builder’s default dialect.
Returns:: A dataclass containing the SQL string and parameters.
Return type:: BuiltQuery

Examples

# Use builder’s default dialect query = sql.select(“*”).from_(“products”) result = query.build()

# Override dialect at build time postgres_sql = query.build(dialect=”postgres”) mysql_sql = query.build(dialect=”mysql”)

to_statement(config=None)[source]¶

Converts the built query into a SQL statement object.

Parameters:: config¶ (StatementConfig | None) – Optional SQL configuration.
Returns:: A SQL statement object.
Return type:: SQL

Tables¶

class sqlspec.builder.CreateTable[source]¶

Bases: DDLBuilder

Builder for CREATE TABLE statements with columns and constraints.

Example

builder = (: CreateTable(“users”) .column(“id”, “SERIAL”, primary_key=True) .column(“email”, “VARCHAR(255)”, not_null=True, unique=True) .column(“created_at”, “TIMESTAMP”, default=”CURRENT_TIMESTAMP”) .foreign_key_constraint(“org_id”, “organizations”, “id”)

) sql = builder.build().sql

__init__(table_name, dialect=None)[source]¶

in_schema(schema_name)[source]¶

Set the schema for the table.

Return type:: Self

if_not_exists()[source]¶

Add IF NOT EXISTS clause.

Return type:: Self

temporary()[source]¶

Create a temporary table.

Return type:: Self

like(source_table)[source]¶

Create table LIKE another table.

Return type:: Self

tablespace(name)[source]¶

Set tablespace for the table.

Return type:: Self

partition_by(partition_spec)[source]¶

Set partitioning specification.

Return type:: Self

property columns: list[ColumnDefinition]¶

Get the list of column definitions for this table.

Returns:: List of ColumnDefinition objects.

column(name, dtype, default=None, not_null=False, primary_key=False, unique=False, auto_increment=False, comment=None, check=None, generated=None, collate=None)[source]¶

Add a column definition to the table.

Return type:: Self

primary_key_constraint(columns, name=None)[source]¶

Add a primary key constraint.

Return type:: Self

foreign_key_constraint(columns, references_table, references_columns, name=None, on_delete=None, on_update=None, deferrable=False, initially_deferred=False)[source]¶

Add a foreign key constraint.

Return type:: Self

unique_constraint(columns, name=None)[source]¶

Add a unique constraint.

Return type:: Self

check_constraint(condition, name=None)[source]¶

Add a check constraint.

Return type:: Self

engine(engine_name)[source]¶

Set storage engine (MySQL/MariaDB).

Return type:: Self

charset(charset_name)[source]¶

Set character set.

Return type:: Self

collate(collation)[source]¶

Set table collation.

Return type:: Self

comment(comment_text)[source]¶

Set table comment.

Return type:: Self

with_option(key, value)[source]¶

Add custom table option.

Return type:: Self

class sqlspec.builder.CreateTableAsSelect[source]¶

Bases: DDLBuilder

Builder for CREATE TABLE [IF NOT EXISTS] … AS SELECT … (CTAS).

Example

builder = (: CreateTableAsSelectBuilder() .name(“my_table”) .if_not_exists() .columns(“id”, “name”) .as_select(select_builder)

) sql = builder.build().sql

- name(table_name: str): Set the table name.

- if_not_exists(): Add IF NOT EXISTS.

- columns(*cols: str): Set explicit column list (optional).

- as_select(select_query): Set the SELECT source (SQL, SelectBuilder, or str).

__init__(dialect=None)[source]¶

class sqlspec.builder.AlterTable[source]¶

Bases: DDLBuilder

Builder for ALTER TABLE operations.

Example

builder = (: AlterTable(“users”) .add_column(“email”, “VARCHAR(255)”, not_null=True) .drop_column(“old_field”) .add_constraint(“check_age”, “CHECK (age >= 18)”)

)

__init__(table_name, dialect=None)[source]¶

if_exists()[source]¶

Add IF EXISTS clause.

Return type:: Self

add_column(name, dtype, default=None, not_null=False, unique=False, comment=None, after=None, first=False)[source]¶

Add a new column to the table.

Return type:: Self

drop_column(name, cascade=False)[source]¶

Drop a column from the table.

Return type:: Self

alter_column_type(name, new_type, using=None)[source]¶

Change the type of an existing column.

Return type:: Self

rename_column(old_name, new_name)[source]¶

Rename a column.

Return type:: Self

add_constraint(constraint_type, columns=None, name=None, references_table=None, references_columns=None, condition=None, on_delete=None, on_update=None)[source]¶

Add a constraint to the table.

Parameters:

constraint_type¶ (str) – Type of constraint (‘PRIMARY KEY’, ‘FOREIGN KEY’, ‘UNIQUE’, ‘CHECK’)
columns¶ (str | list[str] | None) – Column(s) for the constraint (not needed for CHECK)
name¶ (str | None) – Optional constraint name
references_table¶ (str | None) – Table referenced by foreign key
references_columns¶ (str | list[str] | None) – Columns referenced by foreign key
condition¶ (str | ColumnExpression | None) – CHECK constraint condition
on_delete¶ (str | None) – Foreign key ON DELETE action
on_update¶ (str | None) – Foreign key ON UPDATE action

Return type:

Self

drop_constraint(name, cascade=False)[source]¶

Drop a constraint from the table.

Return type:: Self

set_not_null(column)[source]¶

Set a column to NOT NULL.

Return type:: Self

drop_not_null(column)[source]¶

Remove NOT NULL constraint from a column.

Return type:: Self

class sqlspec.builder.DropTable[source]¶

Bases: DDLBuilder

Builder for DROP TABLE [IF EXISTS] … [CASCADE|RESTRICT].

__init__(table_name, dialect=None)[source]¶

Initialize DROP TABLE with table name.

Parameters:

table_name¶ (str) – Name of the table to drop
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.RenameTable[source]¶

Bases: DDLBuilder

Builder for ALTER TABLE … RENAME TO … statements.

__init__(old_name, dialect=None)[source]¶

Initialize RENAME TABLE with old name.

Parameters:

old_name¶ (str) – Current name of the table
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.Truncate[source]¶

Bases: DDLBuilder

Builder for TRUNCATE TABLE … [CASCADE|RESTRICT] [RESTART IDENTITY|CONTINUE IDENTITY].

__init__(table_name, dialect=None)[source]¶

Initialize TRUNCATE with table name.

Parameters:

table_name¶ (str) – Name of the table to truncate
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.CommentOn[source]¶

Bases: DDLBuilder

Builder for COMMENT ON … IS … statements.

__init__(dialect=None)[source]¶

Initialize COMMENT ON builder.

Parameters:: dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

Column and Constraint Definitions¶

class sqlspec.builder._ddl.ColumnDefinition[source]¶

Bases: object

Column definition for CREATE TABLE.

__init__(name, dtype, default=None, not_null=False, primary_key=False, unique=False, auto_increment=False, comment=None, check=None, generated=None, collate=None)[source]¶

class sqlspec.builder._ddl.ConstraintDefinition[source]¶

Bases: object

Constraint definition for CREATE TABLE.

__init__(constraint_type, name=None, columns=None, references_table=None, references_columns=None, condition=None, on_delete=None, on_update=None, deferrable=False, initially_deferred=False)[source]¶

class sqlspec.builder._ddl.AlterOperation[source]¶

Bases: object

Represents a single ALTER TABLE operation.

__init__(operation_type, column_name=None, column_definition=None, constraint_name=None, constraint_definition=None, new_type=None, new_name=None, after_column=None, first=False, using_expression=None)[source]¶

Indexes¶

class sqlspec.builder.CreateIndex[source]¶

Bases: DDLBuilder

Builder for CREATE [UNIQUE] INDEX [IF NOT EXISTS] … ON … (…).

__init__(index_name, dialect=None)[source]¶

Initialize CREATE INDEX with index name.

Parameters:

index_name¶ (str) – Name of the index to create
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.DropIndex[source]¶

Bases: DDLBuilder

Builder for DROP INDEX [IF EXISTS] … [ON table] [CASCADE|RESTRICT].

__init__(index_name, dialect=None)[source]¶

Initialize DROP INDEX with index name.

Parameters:

index_name¶ (str) – Name of the index to drop
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

Views¶

class sqlspec.builder.CreateView[source]¶

Bases: DDLBuilder

Builder for CREATE VIEW [IF NOT EXISTS] … AS SELECT …

__init__(view_name, dialect=None)[source]¶

Initialize CREATE VIEW with view name.

Parameters:

view_name¶ (str) – Name of the view to create
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.CreateMaterializedView[source]¶

Bases: DDLBuilder

Builder for CREATE MATERIALIZED VIEW [IF NOT EXISTS] … AS SELECT …

__init__(view_name, dialect=None)[source]¶

Initialize CREATE MATERIALIZED VIEW with view name.

Parameters:

view_name¶ (str) – Name of the materialized view to create
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.DropView[source]¶

Bases: DDLBuilder

Builder for DROP VIEW [IF EXISTS] … [CASCADE|RESTRICT].

__init__(view_name, dialect=None)[source]¶

Initialize DROP VIEW with view name.

Parameters:

view_name¶ (str) – Name of the view to drop
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.DropMaterializedView[source]¶

Bases: DDLBuilder

Builder for DROP MATERIALIZED VIEW [IF EXISTS] … [CASCADE|RESTRICT].

__init__(view_name, dialect=None)[source]¶

Initialize DROP MATERIALIZED VIEW with view name.

Parameters:

view_name¶ (str) – Name of the materialized view to drop
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

name(view_name)[source]¶

Set the materialized view name.

Return type:: Self

if_exists()[source]¶

Add IF EXISTS clause.

Return type:: Self

cascade()[source]¶

Add CASCADE to drop dependent objects.

Return type:: Self

restrict()[source]¶

Add RESTRICT to prevent dropping if dependencies exist.

Return type:: Self

Schemas¶

class sqlspec.builder.CreateSchema[source]¶

Bases: DDLBuilder

Builder for CREATE SCHEMA [IF NOT EXISTS] schema_name [AUTHORIZATION user_name].

__init__(schema_name, dialect=None)[source]¶

Initialize CREATE SCHEMA with schema name.

Parameters:

schema_name¶ (str) – Name of the schema to create
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use

class sqlspec.builder.DropSchema[source]¶

Bases: DDLBuilder

Builder for DROP SCHEMA [IF EXISTS] … [CASCADE|RESTRICT].

__init__(schema_name, dialect=None)[source]¶

Initialize DROP SCHEMA with schema name.

Parameters:

schema_name¶ (str) – Name of the schema to drop
dialect¶ (Union[str, Dialect, Type[Dialect], None]) – SQL dialect to use