ralsei.connection.jinja

Module Contents

Classes

ConnectionEnvironment

Combines of a database connection with a jinja sql environment,

API

class ralsei.connection.jinja.ConnectionEnvironment(sqlalchemy_conn: sqlalchemy.engine.Engine | ralsei.connection.ext.ConnectionExt, env: ralsei.jinja.ISqlEnvironment | None = None)

Combines of a database connection with a jinja sql environment,

letting you execute dynamically rendered SQL

Parameters:
sqlalchemy_conn: sqlalchemy.engine.Engine | ralsei.connection.ext.ConnectionExt

use exsisting connection or create a new one from engine

env: ralsei.jinja.ISqlEnvironment | None = None

if not provided, a new environment will be created from the engine’s dialect

Initialization

sqlalchemy : ralsei.connection.ext.ConnectionExt = None
jinja : ralsei.jinja.ISqlEnvironment = None
property dialect_info : ralsei.dialect.DialectInfo

Quick access to jinja environment’s DialectInfo

render_execute(source: str, template_params: collections.abc.Mapping[str, Any] = {}, bind_params: sqlalchemy.engine.interfaces._CoreAnyExecuteParams | None = None) sqlalchemy.engine.CursorResult[Any]

Render and execute jinja SQL template

Parameters:
source: str

sql template

template_params: collections.abc.Mapping[str, Any] = {}

jinja template parameters

bind_params: sqlalchemy.engine.interfaces._CoreAnyExecuteParams | None = None

sql bind parameters (see sqlalchemy.engine.Connection.execute() )

render_executescript(source: str | list[str], template_params: collections.abc.Mapping[str, Any] = {}, bind_params: collections.abc.Mapping[str, Any] | None = None)

Render and execute multiple SQL statements

Parameters:
source: str | list[str]

list of sql statements
or a single template with statements separated by {%split%} tag

template_params: collections.abc.Mapping[str, Any] = {}

jinja template parameters

bind_params: collections.abc.Mapping[str, Any] | None = None

sql bind parameters

execute_with_length_hint(statement: sqlalchemy.sql.expression.Executable, parameters: collections.abc.Mapping[str, Any] | None = None) collections.abc.Iterable[sqlalchemy.engine.Row[Any]]

Execute a sql expression, returning an object with a object.__length_hint__() method, letting you see the estimated number or rows.

Concrete implementation depends on the sql dialect

__enter__() ralsei.connection.jinja.ConnectionEnvironment
__exit__(type_: Any, value: Any, traceback: Any) None