ralsei.app¶

Package Contents¶

Classes¶

API¶

class ralsei.app.Ralsei(url: sqlalchemy.engine.URL, pipeline: ralsei.graph.Pipeline)¶

The pipeline-running CLI application

Decorate your subclass with click.option() decorator to add custom CLI options. Positional arguments are not allowed

Parameters:¶

url: sqlalchemy.engine.URL¶

When the class constructor is called by the CLI, the URL is provided as the first argument

pipeline: ralsei.graph.Pipeline¶

The CLI does not give you the pipeline, you must create one in your subclass and pass it to super().__init__()

Example

@click.option("-s", "--schema", help="Database schema")
class App(Ralsei):
    def __init__(
        url: sqlalchemy.URL, # First argument must always be the url
        schema: str | None, # Custom argument added with the click decorator
    ):
        super().__init__(url, MyPipeline(schema))

if __name__ == "__main__":
    App.run_cli()

Initialization

pipeline : ralsei.graph.Pipeline = None¶

engine : sqlalchemy.engine.Engine = None¶

env : ralsei.jinja.SqlEnvironment = None¶

dag : ralsei.graph.DAG = None¶

_create_engine(url: sqlalchemy.engine.URL) → sqlalchemy.engine.Engine¶

Override this to customize engine creation

_prepare_env(env: ralsei.jinja.SqlEnvironment)¶

Here you can add your own filters/globals to the jinja environment

connect() → ralsei.connection.ConnectionEnvironment¶

Creates a new connection, returns connection + jinja env

_on_connect(conn: ralsei.connection.ConnectionEnvironment)¶

Run custom code after database connection

classmethod build_cli() → click.Group¶

Create a click CLI based on this class

classmethod run_cli(*args, **kwargs)¶

Build and run click CLI, print traceback in case of exception