ralsei.task.create_table¶

Module Contents¶

Classes¶

API¶

class ralsei.task.create_table.CreateTableTask¶

Bases: ralsei.task.base.TaskImpl

Base class for a task that performs table creation

All you have to do is call _prepare_table() from within ralsei.task.TaskImpl.prepare().

output, _exists() and _delete() are implemented for you, leaving only the ralsei.task.TaskImpl._run() part

Example

import pandas as pd

class UploadCsv(TaskDef):
    table: Table
    path: Path

    class Impl(CreateTableTask):
        def prepare(self, this: "UploadCsv"):
            self._prepare_table(this.table)
            self.__path = this.path

        def _run(self, conn: ConnectionEnvironment):
            with self.__path.open() as file:
                pd.read_csv(file).to_sql(
                    self._table.name,
                    conn.sqlalchemy,
                    schema=self._table.schema
                )

_table : ralsei.types.Table = None¶

_drop_sql : sqlalchemy.sql.elements.TextClause = None¶

_prepare_table(table: ralsei.types.Table, view: bool = False)¶

property output : Any¶

_exists(conn: ralsei.connection.ConnectionEnvironment) → bool¶

_delete(conn: ralsei.connection.ConnectionEnvironment)¶