Compilers

pelican.compilers.compiler.DialectCompiler

DialectCompiler(engine)

Bases: ABC

Source code in pelican/compilers/compiler.py

def __init__(self, engine: Engine) -> None:
    self.engine = engine
    self.dialect = engine.dialect

engine instance-attribute

engine = engine

dialect instance-attribute

dialect = dialect

add_column

add_column(table_name, column)

Source code in pelican/compilers/compiler.py

def add_column(self, table_name: str, column: Column) -> Iterable[Executable]:
    sql = f"ALTER TABLE {table_name} ADD COLUMN {CreateColumn(column).compile(dialect=self.dialect)}"
    return [text(sql)]

drop_column

drop_column(table_name, column_name)

Source code in pelican/compilers/compiler.py

def drop_column(self, table_name: str, column_name: str) -> Iterable[Executable]:
    sql = f"ALTER TABLE {table_name} DROP COLUMN {column_name}"
    return [text(sql)]

rename_column abstractmethod

rename_column(table_name, old_name, new_name)

Source code in pelican/compilers/compiler.py

@abstractmethod
def rename_column(
    self, table_name: str, old_name: str, new_name: str
) -> Iterable[DDL]:
    pass

alter_column abstractmethod

alter_column(table_name, column_name, new_type=None, nullable=None, default=None, server_default=None)

Source code in pelican/compilers/compiler.py

@abstractmethod
def alter_column(
    self,
    table_name: str,
    column_name: str,
    new_type: TypeEngine | None = None,
    nullable: bool | None = None,
    default: Any = None,
    server_default: Any = None,
) -> Iterable[DDL]:
    pass

create_index

create_index(table_name, index_name, column_names, unique=False)

Source code in pelican/compilers/compiler.py

def create_index(
    self,
    table_name: str,
    index_name: str,
    column_names: list[str],
    unique: bool = False,
) -> Iterable[DDLElement]:
    metadata = MetaData()
    table = Table(
        table_name, metadata, autoload_with=self.engine, extend_existing=True
    )

    columns = [table.c[col_name] for col_name in column_names]
    index = Index(index_name, *columns, unique=unique)

    return [CreateIndex(index)]

drop_index

drop_index(table_name, index_name)

Source code in pelican/compilers/compiler.py

def drop_index(self, table_name: str, index_name: str) -> Iterable[DDLElement]:
    metadata = MetaData()
    table = Table(table_name, metadata)
    index = Index(index_name, _table=table)

    return [DropIndex(index)]

pelican.compilers.sqlite.SQLiteCompiler

SQLiteCompiler(engine)

Bases: DialectCompiler

Source code in pelican/compilers/compiler.py

def __init__(self, engine: Engine) -> None:
    self.engine = engine
    self.dialect = engine.dialect

rename_column

rename_column(table_name, old_name, new_name)

Source code in pelican/compilers/sqlite.py

def rename_column(
    self, table_name: str, old_name: str, new_name: str
) -> Iterable[DDL]:
    return [DDL(f"ALTER TABLE {table_name} RENAME COLUMN {old_name} TO {new_name}")]

alter_column

alter_column(table_name, column_name, new_type=None, nullable=None, default=None, server_default=None)

SQLite doesn't support ALTER COLUMN

Source code in pelican/compilers/sqlite.py

def alter_column(
    self,
    table_name: str,
    column_name: str,
    new_type: TypeEngine | None = None,
    nullable: bool | None = None,
    default: Any = None,
    server_default: Any = None,
) -> Iterable[DDL]:
    """SQLite doesn't support ALTER COLUMN"""
    raise NotImplementedError(
        f"SQLite does not support ALTER COLUMN for '{column_name}'. "
        "Column type/constraint changes require table recreation. "
        "Use batch_alter_table() instead."
    )

pelican.compilers.postgresql.PostgreSQLCompiler

PostgreSQLCompiler(engine)

Bases: DialectCompiler

Source code in pelican/compilers/compiler.py

def __init__(self, engine: Engine) -> None:
    self.engine = engine
    self.dialect = engine.dialect

rename_column

rename_column(table_name, old_name, new_name)

Source code in pelican/compilers/postgresql.py

def rename_column(
    self, table_name: str, old_name: str, new_name: str
) -> Iterable[DDL]:
    return [DDL(f"ALTER TABLE {table_name} RENAME COLUMN {old_name} TO {new_name}")]

alter_column

alter_column(table_name, column_name, new_type=None, nullable=None, default=None, server_default=None)

Source code in pelican/compilers/postgresql.py

def alter_column(
    self,
    table_name: str,
    column_name: str,
    new_type: TypeEngine | None = None,
    nullable: bool | None = None,
    default: Any = None,
    server_default: Any = None,
) -> Iterable[DDL]:
    statements = []

    if new_type is not None:
        type_str = self.dialect.type_compiler_instance.process(new_type)
        statements.append(
            DDL(
                f"ALTER TABLE {table_name} ALTER COLUMN {column_name} TYPE {type_str}"
            )
        )

    if nullable is True:
        statements.append(
            DDL(
                f"ALTER TABLE {table_name} ALTER COLUMN {column_name} DROP NOT NULL"
            )
        )
    elif nullable is False:
        statements.append(
            DDL(f"ALTER TABLE {table_name} ALTER COLUMN {column_name} SET NOT NULL")
        )

    if server_default is not None:
        statements.append(
            DDL(
                f"ALTER TABLE {table_name} ALTER COLUMN {column_name} SET DEFAULT {server_default}"
            )
        )

    if not statements:
        raise ValueError(
            f"alter_column for '{column_name}' requires at least one change "
            "(new_type, nullable, or server_default)"
        )

    return statements