`ralsei.connection`¶

Package Contents¶

`ConnectionExt`	Extends sqlalchemy’s Connection with additional utility methods
`ConnectionEnvironment`	Combines of a database connection with a jinja sql environment,

`create_engine`	Wrapper around `sqlalchemy.create_engine()`

ralsei.connection.create_engine(url: str | sqlalchemy.engine.URL, **kwargs) → sqlalchemy.engine.Engine¶

Applies additional configurations for sqlite, such as enabling foreign_keys and fixing transaction issues (https://docs.sqlalchemy.org/en/20/dialects/sqlite.html#serializable-isolation-savepoints-transactional-ddl)

class ralsei.connection.ConnectionExt(engine: sqlalchemy.engine.base.Engine, connection: sqlalchemy.pool.PoolProxiedConnection | None = None, _has_events: bool | None = None, _allow_revalidate: bool = True, _allow_autobegin: bool = True)¶

Extends sqlalchemy’s Connection with additional utility methods

Initialization

Construct a new Connection.

execute_text(statement: str, parameters: sqlalchemy.engine.interfaces._CoreAnyExecuteParams | None = None) → sqlalchemy.engine.CursorResult[Any]¶: Execute a sql string

executescript(statements: collections.abc.Iterable[sqlalchemy.sql.expression.Executable], parameters: collections.abc.Mapping[str, Any] | None = None)¶: Execute a series of statements, similar to sqlite’s executescript

executescript_text(statements: collections.abc.Iterable[str], parameters: collections.abc.Mapping[str, Any] | None = None)¶: Execute a series of string statements, similar to sqlite’s executescript

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

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