Skip to content

Compilers

pelican.compilers.compiler.DialectCompiler 

DialectCompiler(engine)

Bases: ABC

Source code in pelican/compilers/compiler.py 
23
24
25
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 
27
28
29
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 
31
32
33
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 
35
36
37
38
39
@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 
41
42
43
44
45
46
47
48
49
50
51
@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 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
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 
70
71
72
73
74
75
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)]

add_foreign_key 

add_foreign_key(table_name, columns, ref_table, ref_columns, name=None, on_delete=None)
Source code in pelican/compilers/compiler.py 
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def add_foreign_key(
    self,
    table_name: str,
    columns: list[str],
    ref_table: str,
    ref_columns: list[str],
    name: str | None = None,
    on_delete: str | None = None,
) -> Iterable[DDL]:
    cols = ", ".join(columns)
    ref_cols = ", ".join(ref_columns)
    constraint = f"CONSTRAINT {name} " if name else ""
    on_delete_clause = f" ON DELETE {on_delete}" if on_delete else ""
    return [
        DDL(
            f"ALTER TABLE {table_name} ADD {constraint}"
            f"FOREIGN KEY ({cols}) REFERENCES {ref_table} ({ref_cols}){on_delete_clause}"
        )
    ]

drop_foreign_key 

drop_foreign_key(table_name, constraint_name)
Source code in pelican/compilers/compiler.py 
97
98
def drop_foreign_key(self, table_name: str, constraint_name: str) -> Iterable[DDL]:
    return [DDL(f"ALTER TABLE {table_name} DROP CONSTRAINT {constraint_name}")]

pelican.compilers.sqlite.SQLiteCompiler 

SQLiteCompiler(engine)

Bases: DialectCompiler

Source code in pelican/compilers/compiler.py 
23
24
25
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 
 9
10
11
12
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/sqlite.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
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]:
    raise UnsupportedDialectError(
        f"SQLite does not support ALTER COLUMN for '{column_name}'. "
        "Column type/constraint changes require table recreation."
    )

add_foreign_key 

add_foreign_key(table_name, columns, ref_table, ref_columns, name=None, on_delete=None)
Source code in pelican/compilers/sqlite.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
def add_foreign_key(
    self,
    table_name: str,
    columns: list[str],
    ref_table: str,
    ref_columns: list[str],
    name: str | None = None,
    on_delete: str | None = None,
) -> Iterable[DDL]:
    raise UnsupportedDialectError(
        f"SQLite does not support ADD CONSTRAINT FOREIGN KEY on '{table_name}'. "
        "Foreign key changes require table recreation."
    )

drop_foreign_key 

drop_foreign_key(table_name, constraint_name)
Source code in pelican/compilers/sqlite.py 
42
43
44
45
46
def drop_foreign_key(self, table_name: str, constraint_name: str) -> Iterable[DDL]:
    raise UnsupportedDialectError(
        f"SQLite does not support DROP CONSTRAINT on '{table_name}'. "
        "Foreign key changes require table recreation."
    )

pelican.compilers.postgresql.PostgreSQLCompiler 

PostgreSQLCompiler(engine)

Bases: DialectCompiler

Source code in pelican/compilers/compiler.py 
23
24
25
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 
10
11
12
13
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 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
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