`ralsei.task.base`¶

Module Contents¶

Classes¶

`Task`	Base task class
`TaskImpl`	Task implementation created from `TaskDef` arguments
`TaskDef`	Stores task aguments before said task is created

API¶

class ralsei.task.base.Task¶

Bases: abc.ABC

Base task class

abstract run(conn: ralsei.connection.ConnectionExt)¶: Run the task

abstract delete(conn: ralsei.connection.ConnectionExt)¶: Delete whatever run() has created

redo(conn: ralsei.connection.ConnectionExt)¶: Calls delete() and then run()

abstract property output : Any¶

Object created or modified by this task (usually a ralsei.types.Table)

Used for resolving ralsei.graph.Pipeline.outputof()

abstract exists(conn: ralsei.connection.ConnectionExt) → bool¶: Check if task has already been done

scripts() → collections.abc.Iterable[tuple[str, object]]¶

Get SQL scripts rendered by this task

Returns:¶

iterable of ("name", script), where script is either:

a string-like object, usually str or sqlalchemy.sql.elements.TextClause
a list of string-like objects (in case of multiple statements)

class ralsei.task.base.TaskImpl(this: D, env: ralsei.jinja.ISqlEnvironment)¶

Bases: ralsei.task.base.Task

Task implementation created from TaskDef arguments

Parameters:¶

this : TaskDef¶: the settings object for this task
env: ralsei.jinja.ISqlEnvironment¶: jinja environment

Warning

It is advised againts overriding __init__. Perform your initialization in prepare() instead.

Initialization

env : ralsei.jinja.ISqlEnvironment = None¶

_scripts : dict[str, object] = None¶

You can save your sql scripts here when you render them, the key-value pairs will be returned by scripts()

Example

class Impl(TaskImpl)
    def prepare(self, this: "MyTaskDef")
        self._scripts["Create table"] = self.__create = self.env.render(this.sql)

prepare(this: D)¶

Perform your initialization here

Parameters:¶

this : TaskDef¶: the settings object for this task

resolve(value: ralsei.graph.Resolves[ralsei.task.base.TaskImpl.resolve.T]) → ralsei.task.base.TaskImpl.resolve.T¶

Resolve a dependency

Parameters:¶

value : ralsei.graph.OutputOf | T¶: may or may not need dependency resolution

Returns:¶

the resolved value

Return type:¶

run(conn: ralsei.connection.ConnectionExt)¶

delete(conn: ralsei.connection.ConnectionExt)¶

exists(conn: ralsei.connection.ConnectionExt) → bool¶

abstract _run(conn: ralsei.connection.ConnectionEnvironment)¶: Run the task

abstract _delete(conn: ralsei.connection.ConnectionEnvironment)¶: Delete whatever _run() has created

abstract _exists(conn: ralsei.connection.ConnectionEnvironment) → bool¶: Check if task has already been done

scripts() → collections.abc.Iterable[tuple[str, object]]¶: Get SQL scripts rendered by this task

class ralsei.task.base.TaskDef¶

Stores task aguments before said task is created

Any subclass of TaskDef automatically gets dataclasses.dataclass() decorator applied to it

Impl : ClassVar[type[ralsei.task.base.TaskImpl[Self]]] = None¶

The associated task class

Note

This field is not part of the dataclass

Example

class MyTask(TaskDef):
    class Impl(TaskImpl):
        def prepare(self, this: "MyTask"):
            ...

locals : dict[str, Any] = 'field(...)'¶: Local variables added to the jinja environment

create(env: ralsei.jinja.SqlEnvironment) → ralsei.task.base.TaskImpl[Self]¶: Instantiate the associated Impl

ralsei.task.base¶

Module Contents¶

Classes¶

API¶

`ralsei.task.base`¶