Skip to content

Schema

pelican.schema.helpers.TableBuilder 

TableBuilder(table_name, metadata, primary_key=True, table=None)
Source code in pelican/schema/helpers.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
def __init__(
    self,
    table_name: str,
    metadata: MetaData,
    primary_key: bool = True,
    table: Table | None = None,
) -> None:
    self.table_name = table_name
    self.metadata = metadata
    self.table = table if table is not None else Table(self.table_name, metadata)

    self._is_existing_table = table is not None
    self.operations: list[Operation] = []

    if primary_key and not self._is_existing_table:
        self.integer("id", primary_key=True, autoincrement=True)

table_name instance-attribute 

table_name = table_name

metadata instance-attribute 

metadata = metadata

table instance-attribute 

table = table if table is not None else Table(table_name, metadata)

operations instance-attribute 

operations = []

column 

column(name, type_, *args, **kwargs)
Source code in pelican/schema/helpers.py 
49
50
51
52
53
54
55
def column(self, name: str, type_: _T, *args: Any, **kwargs: Any) -> None:
    column_ = Column(name, type_, *args, **kwargs)
    # TODO: Add Column class for the builder which would then be used to build the SA Column in the compiler
    self.table.append_column(column_, replace_existing=True)

    if self._is_existing_table:
        self.operations.append(AddColumn(self.table_name, column_))

alter 

alter(name, **kwargs)
Source code in pelican/schema/helpers.py 
57
58
59
60
def alter(self, name: str, **kwargs: Any) -> None:
    if not self._is_existing_table:
        raise ValueError("alter can only be used on existing table")
    self.operations.append(AlterColumn(self.table_name, name, **kwargs))

rename 

rename(old_name, new_name)
Source code in pelican/schema/helpers.py 
62
63
64
65
def rename(self, old_name: str, new_name: str) -> None:
    if not self._is_existing_table:
        raise ValueError("rename can only be used on existing table")
    self.operations.append(RenameColumn(self.table_name, old_name, new_name))

drop 

drop(name)
Source code in pelican/schema/helpers.py 
67
68
69
70
def drop(self, name: str) -> None:
    if not self._is_existing_table:
        raise ValueError("drop can only be used on existing table")
    self.operations.append(DropColumn(self.table_name, name))

integer 

integer(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
72
73
def integer(self, name: str, *args: Any, **kwargs: Any) -> None:
    self.column(name, Integer, *args, **kwargs)

float 

float(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
75
76
def float(self, name: str, *args: Any, **kwargs: Any) -> None:
    self.column(name, Float, *args, **kwargs)

double 

double(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
78
79
def double(self, name: str, *args: Any, **kwargs: Any) -> None:
    self.column(name, Double, *args, **kwargs)

boolean 

boolean(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
81
82
def boolean(self, name: str, *args: Any, **kwargs: Any) -> None:
    self.column(name, Boolean, *args, **kwargs)

string 

string(name, length=255, *args, **kwargs)
Source code in pelican/schema/helpers.py 
84
85
def string(self, name: str, length: int = 255, *args: Any, **kwargs: Any) -> None:
    self.column(name, String(length), *args, **kwargs)

text 

text(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
87
88
def text(self, name: str, *args: Any, **kwargs: Any) -> None:
    self.column(name, Text, *args, **kwargs)

datetime 

datetime(name, *args, **kwargs)
Source code in pelican/schema/helpers.py 
90
91
92
def datetime(self, name: str, *args: Any, **kwargs: Any) -> None:
    default = kwargs.pop("default", func.now())
    self.column(name, DateTime, default=default, *args, **kwargs)

timestamps 

timestamps()
Source code in pelican/schema/helpers.py 
94
95
96
def timestamps(self) -> None:
    self.datetime("created_at", nullable=False)
    self.datetime("updated_at", onupdate=func.now(), nullable=False)

references 

references(model_name, on_delete='CASCADE', **kwargs)
Source code in pelican/schema/helpers.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def references(
    self, model_name: str, on_delete: str = "CASCADE", **kwargs: Any
) -> None:
    self.table.append_column(
        Column(
            f"{model_name}_id",
            Integer,
            ForeignKey(
                f"{inflection.pluralize(model_name)}.id", ondelete=on_delete
            ),
            **kwargs,
        )
    )

index 

index(column_names, *, name=None, unique=False)
Source code in pelican/schema/helpers.py 
112
113
114
115
116
117
118
119
120
121
122
123
def index(
    self, column_names: list[str], *, name: str | None = None, unique: bool = False
) -> None:
    if not column_names:
        raise ValueError("At least one column name is required for an index")

    if name is None:
        name = f"{self.table_name}_{'_'.join(column_names)}_idx"

    self.operations.append(
        CreateIndex(self.table_name, name, column_names, unique=unique)
    )

remove_index 

remove_index(column_names=None, *, name=None)
Source code in pelican/schema/helpers.py 
125
126
127
128
129
130
131
132
133
134
135
136
def remove_index(
    self, column_names: list[str] | None = None, *, name: str | None = None
) -> None:
    if not self._is_existing_table:
        raise ValueError("remove_index can only be used on existing table")

    if not name:
        if not column_names:
            raise ValueError("At least one column name is required for an index")
        name = f"{self.table_name}_{'_'.join(column_names)}_idx"

    self.operations.append(RemoveIndex(self.table_name, name))

pelican.schema.helpers.create_table 

create_table(table_name, primary_key=True)

Create a new table

Example

from pelican import create_table


@migration.up()
def upgrade():
    with create_table('spaceships') as t:
        t.string('name', nullable=False)
        t.string('designation', nullable=False)
        t.integer('crew_capacity', default=1)
        t.timestamps()
Source code in pelican/schema/helpers.py 
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
@contextmanager
def create_table(table_name: str, primary_key: bool = True) -> Iterator[TableBuilder]:
    """Create a new table

    ## Example

    ```python
    from pelican import create_table


    @migration.up()
    def upgrade():
        with create_table('spaceships') as t:
            t.string('name', nullable=False)
            t.string('designation', nullable=False)
            t.integer('crew_capacity', default=1)
            t.timestamps()
    ```
    """
    from pelican import runner

    builder = TableBuilder(table_name, runner.metadata, primary_key=primary_key)
    yield builder

    with runner.engine.begin() as conn:
        builder.table.create(conn, checkfirst=True)

    runner.execute_operations(builder.operations)

pelican.schema.helpers.change_table 

change_table(table_name)

Modify an existing table

Example

from pelican import change_table


@migration.up()
def upgrade():
    with change_table('spaceships') as t:
        t.string('name', nullable=False) # add column
        t.alter('name', nullable=True) # alter column
        t.rename('name', 'new_name') # rename column
        t.drop('new_name') # drop column
Source code in pelican/schema/helpers.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
@contextmanager
def change_table(table_name: str) -> Iterator[TableBuilder]:
    """Modify an existing table

    ## Example

    ```python
    from pelican import change_table


    @migration.up()
    def upgrade():
        with change_table('spaceships') as t:
            t.string('name', nullable=False) # add column
            t.alter('name', nullable=True) # alter column
            t.rename('name', 'new_name') # rename column
            t.drop('new_name') # drop column
    ```
    """
    from pelican import runner

    table = Table(
        table_name, runner.metadata, autoload_with=runner.engine, extend_existing=True
    )

    builder = TableBuilder(table_name, runner.metadata, table=table)
    yield builder

    runner.execute_operations(builder.operations)

pelican.schema.helpers.drop_table 

drop_table(table_name)

Drop an existing table

Example

from pelican import drop_table


@migration.down()
def downgrade():
    drop_table('spaceships')
Source code in pelican/schema/helpers.py 
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
def drop_table(table_name: str) -> None:
    """Drop an existing table

    ## Example

    ```python
    from pelican import drop_table


    @migration.down()
    def downgrade():
        drop_table('spaceships')
    ```
    """
    from pelican import runner

    with runner.engine.begin() as conn:
        table = Table(table_name, runner.metadata, autoload_with=runner.engine)
        table.drop(conn)