Schema

pelican.schema.helpers.TableBuilder

TableBuilder(table_name, metadata, primary_key=True, table=None)

Source code in pelican/schema/helpers.py

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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)
    )

add_foreign_key

add_foreign_key(columns, ref_table, ref_columns, *, name=None, on_delete=None)

Source code in pelican/schema/helpers.py

def add_foreign_key(
    self,
    columns: list[str],
    ref_table: str,
    ref_columns: list[str],
    *,
    name: str | None = None,
    on_delete: str | None = None,
) -> None:
    self.operations.append(
        AddForeignKeyConstraint(
            self.table_name,
            columns=columns,
            ref_table=ref_table,
            ref_columns=ref_columns,
            constraint_name=name,
            on_delete=on_delete,
        )
    )

remove_foreign_key

remove_foreign_key(*, name)

Source code in pelican/schema/helpers.py

def remove_foreign_key(self, *, name: str) -> None:
    self.operations.append(DropForeignKeyConstraint(self.table_name, name))

remove_index

remove_index(column_names=None, *, name=None)

Source code in pelican/schema/helpers.py

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

@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()
    ```
    """
    runner = get_runner()
    builder = TableBuilder(table_name, runner.metadata, primary_key=primary_key)
    yield builder

    with runner.engine.connect() as conn:
        with conn.begin():
            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

@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
    ```
    """
    runner = get_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

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')
    ```
    """
    runner = get_runner()

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