Package sqlax

SQLAlchemy Extensions

Download python-snippets-0.2.zip

Provides modules

• sql_statements (user API)
• sql_statements_test (extensive tests)

Compatible with SQLAlchemy v0.7.9 - v1.1.x.

Compatible with Python 2 and Python 3.

Module sql_statements

Generate SQL statements from SQLAlchemy queries, INSERT and UPDATE statements.

User API:

• table_dump_sql()
• sql_statement()
• fix_null_params()
• fix_null_values()
• null_if_none()
• LiteralDialect
• COMPILE_KWARGS

Some code incorporated from python - SQLAlchemy: print the actual query - Stack Overflow.

Automatic Exports

>>> for ex in __all__: printf(sformat('from {0} import {1}', __name__, ex))
from sql_statements import table_dump_sql
from sql_statements import sql_statement
from sql_statements import fix_null_params
from sql_statements import fix_null_values
from sql_statements import null_if_none
from sql_statements import LiteralDialect
from sql_statements import COMPILE_KWARGS

Explicit Exports

>>> if '__all_internal__' in globals():
...   for ex in __all_internal__:
...     printf(sformat('from {0} import {1}', __name__, ex))
from sql_statements import StringLiteral
from sql_statements import get_compile_kwargs
from sql_statements import literal_pre_process

Details

sql_statements.table_dump_sql(table, rows, columns=None, update=None, no_create=None, dialect=None)

Dump sqlalchemy table and data rows as SQL CREATE TABLE and INSERT/UPDATE statements..

Parameters:

• table – SQLAlchemy Table or ORM table
• rows – list of row data lists
• columns – list of column names matching the rows
• update – generate UPDATE statements instead of INSERT.
• no_create – do not generate CREATE TABLE DDL expression.
• dialect – defaults to LiteralDialect.

Handles both SQL tables and ORM tables transparently.

See Test Environment for defintions.

>>> from sqlalchemy import select, insert, update
>>> from sql_statements_test import *

Check SQL table

>>> print(nts(table_dump_sql(TestTable, (TEST_TABLE_DATA, )))) 
CREATE TABLE db_table (
    mycol INTEGER NOT NULL...,
    cboolean BOOL...,
    cinteger INTEGER,
    cfloat FLOAT,
    cdate DATE,
    ctime TIME,
    cdatetime DATETIME,
    cstring VARCHAR(50),
    ctext TEXT,
    cnull INTEGER,
    cenum...,
    PRIMARY KEY (mycol),
    CHECK (cboolean IN (0, 1))...
);
INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext, cnull, cenum) VALUES (5, ..., 100000000000000000000, 3.14159, '2016-10-03', '13:45:00', '2015-06-24 18:09:29.042517', 'UTF-8 snowman: ☃', 'snowman: ☃', NULL, 'foo');

Check ORM table

>>> print(table_dump_sql(TestTableORM, [[1, 10 ** 20, '5'], [2, 6, 'text'], [3, 9, 'done']], 'a b c'.split())) 
CREATE TABLE db_table_orm (
    b INTEGER,
    c VARCHAR(10),
    a INTEGER NOT NULL...,
    PRIMARY KEY (a)
);
INSERT INTO db_table_orm (b, c, a) VALUES (100000000000000000000, '5', 1);
INSERT INTO db_table_orm (b, c, a) VALUES (6, 'text', 2);
INSERT INTO db_table_orm (b, c, a) VALUES (9, 'done', 3);

Check Update on ORM table

>>> print(table_dump_sql(TestTableORM, [[1, 10 ** 20, '5'], [2, 6, 'text'], [3, 9, 'done']], 'a b c'.split(), update=True, no_create=True))
UPDATE db_table_orm SET b=100000000000000000000, c='5', a=1 WHERE db_table_orm.a = 1;
UPDATE db_table_orm SET b=6, c='text', a=2 WHERE db_table_orm.a = 2;
UPDATE db_table_orm SET b=9, c='done', a=3 WHERE db_table_orm.a = 3;

sql_statements.sql_statement(stmt, dialect=None)

Render SQL for a SQA statement.

Returns:

SQL string.

Parameters:

• stmt – SQLAlchemy ORM query or select, insert, update expression.
• dialect – defaults to LiteralDialect.

See Test Environment for defintions.

Check correct rendering of various data types

>>> from sqlalchemy import select, insert, update
>>> from sql_statements_test import *

>>> statement = select([TestTable]).where(TestTable.c.mycol.in_(TEST_TABLE_DATA)).limit(1)
>>> print(nts(sql_statement(statement))) 
SELECT db_table.mycol, db_table.cboolean, db_table.cinteger, db_table.cfloat, db_table.cdate, db_table.ctime, db_table.cdatetime, db_table.cstring, db_table.ctext, db_table.cnull, db_table.cenum
FROM db_table
WHERE db_table.mycol IN (5, 1, 100000000000000000000, 3.14159, '2016-10-03', '13:45:00', '2015-06-24 18:09:29.042517', 'UTF-8 snowman: ☃', 'snowman: ☃', NULL, 'foo')
 LIMIT 1...;

Check correct rendering of INSERT, UPDATE

>>> insert_stmt = insert(TestTableORM.__table__).values(a=1, b=2)
>>> print(sql_statement(insert_stmt))
INSERT INTO db_table_orm (b, a) VALUES (2, 1);

>>> update_stmt = update(TestTableORM.__table__).where(TestTableORM.a == 1).values(b=2)
>>> print(sql_statement(update_stmt))
UPDATE db_table_orm SET b=2 WHERE db_table_orm.a = 1;

See also python - SQLAlchemy: print the actual query - Stack Overflow and python - Sqlalchemy - how to get raw sql from insert(), update() statements with binded params? - Stack Overflow.

sql_statements.fix_null_params(params)

Convert None values to sqlalchemy.null instances.

Parameters:params – list of (column, value) tuples or dict().

sql_statements.fix_null_values(values)

Convert None values to sqlalchemy.null instances.

sql_statements.null_if_none(value)

If value is None, return sqlalchemy.null instance.

class sql_statements.LiteralDialect(**kwargs)

Teach SA how to literalize various things (LiteralDialect).

See python - SQLAlchemy: print the actual query - Stack Overflow

sql_statements.COMPILE_KWARGS = {'compile_kwargs': {'literal_binds': True}}

Keyword arguments for statement.compile as determined by get_compile_kwargs().

If statement.compile does not support keyword argument compile_kwargs, this is an empty dictionary. Otherwise it is a defined as:

{'compile_kwargs': { 'literal_binds': True}}

class sql_statements.StringLiteral(*args, **kwargs)

Teach SA how to literalize various things (StringLiteral).

See python - SQLAlchemy: print the actual query - Stack Overflow.

literal_processor(dialect)

Return a conversion function for processing literal values that are to be rendered directly without using binds.

This function is used when the compiler makes use of the “literal_binds” flag, typically used in DDL generation as well as in certain scenarios where backends don’t accept bound parameters.

New in version 0.9.0.

sql_statements.get_compile_kwargs()

Determine, whether statement.compile supports keyword argument compile_kwargs.

Called once to define COMPILE_KWARGS.

sql_statements.literal_pre_process(value, type_, dialect)

Returns:

Module sql_statements_test

Tests for features of sql_statements.

The rationale for the sql_statements module is documented in test_direct_compilation().

Other dummy functions defined for doctest provide regression rests for sql_statements (they are actually duplicated there):

• test_data_types()
• test_insert_update()
• test_table_dump()

Automatic Exports

>>> for ex in __all__: printf(sformat('from {0} import {1}', __name__, ex))
from sql_statements_test import TestTable
from sql_statements_test import TEST_TABLE_DATA
from sql_statements_test import TestTableORM

Explicit Exports

>>> if '__all_internal__' in globals():
...   for ex in __all_internal__:
...     printf(sformat('from {0} import {1}', __name__, ex))
from sql_statements_test import test_direct_compilation
from sql_statements_test import test_data_types
from sql_statements_test import test_insert_update
from sql_statements_test import test_table_dump

Test Environment

import datetime
from decimal import Decimal

from sqlalchemy import Table, Column, Boolean, Integer, Float, Date, Time, DateTime, String, Text, Enum
from sqlalchemy import select, insert, update

from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base() # use common metadata

TestTable = Table(
    'db_table', Base.metadata,
    Column('mycol', Integer(), primary_key=True),
    Column('cboolean', Boolean()),
    Column('cinteger', Integer()),
    Column('cfloat', Float()),
    Column('cdate', Date()),
    Column('ctime', Time()),
    Column('cdatetime', DateTime()),
    Column('cstring', String(50)),
    Column('ctext', Text()),
    Column('cnull', Integer()),
    Column('cenum', Enum("foo", "bar", "baz")),
)

TEST_TABLE_DATA = (
    5,
    1,
    10 ** 20,  # a long integer
    Decimal('3.14159'),
    datetime.date(2016, 10, 3),
    datetime.time(13, 45, 0),
    datetime.datetime(2015, 6, 24, 18, 9, 29, 42517),
    u8s('UTF-8 ' + 'snowman: ☃'),
    ucs('snowman: ☃'),
    None,
    "foo",
)

class TestTableORM(Base):
    __tablename__ = "db_table_orm"

    b = Column(Integer)
    c = Column(String(10))
    a = Column(Integer, primary_key=True)

Details

sql_statements_test.test_direct_compilation()

Direct compilation of statements.

The resulting feature matrix is:

SQA version	compile_kwargs	SELECT	INSERT/UPDATE	datetime
0.7.9	–	–	–	–
0.9.4	√	1/2	–	–
1.0.11	√	√	–	–
1.0.13	√	√	1/2	–
1.1.6	√	√	1/2	–

Support for keyword compile_kwargs:

version	compile_kwargs
0.7.9	{}
0.9.4	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.11	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.13	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.6	{‘compile_kwargs’: {‘literal_binds’: True}}

See Test Environment for defintions.

SQLAlchemy v1.0.11+ actually substitutes simple bind parameters for queries (SELECT) with any dialect:

>>> from sqlalchemy.dialects import mysql
>>> dialect = mysql.dialect()

>>> select_stmt = select([TestTableORM.__table__]).where(TestTableORM.a.in_((1, 2, ))).limit(1)
>>> printf(nts(select_stmt.compile(dialect=dialect, **COMPILE_KWARGS))) # 0 
SELECT db_table_orm.b, db_table_orm.c, db_table_orm.a
FROM db_table_orm
WHERE db_table_orm.a IN (1, 2)
 LIMIT 1

0.7.9	-	... WHERE db_table_orm.a IN (%s, %s) LIMIT(%s)
0.9.4	1/2	... WHERE db_table_orm.a IN (1, 2) LIMIT(%s)
1.0.11	√	... WHERE db_table_orm.a IN (1, 2) LIMIT(1)
1.0.13	√	... WHERE db_table_orm.a IN (1, 2) LIMIT(1)
1.1.6	√	... WHERE db_table_orm.a IN (1, 2) LIMIT(1)

SQLAlchemy v1.0.13+ also substitutes simple bind parameters for INSERT and UPDATE:

>>> insert_stmt = insert(TestTableORM.__table__).values(a=1, c='1')
>>> printf(insert_stmt.compile(dialect=dialect, **COMPILE_KWARGS)) # 1
INSERT INTO db_table_orm (c, a) VALUES (%s, %s)

0.7.9	-	INSERT INTO db_table_orm (b, a) VALUES (%s, %s)
0.9.4	-	INSERT INTO db_table_orm (b, a) VALUES (%s, %s)
1.0.11	-	INSERT INTO db_table_orm (b, a) VALUES (%s, %s)
1.0.13	√	INSERT INTO db_table_orm (c, a) VALUES ('1', 1)
1.1.6	√	INSERT INTO db_table_orm (c, a) VALUES ('1', 1)

>>> update_stmt = update(TestTableORM.__table__).where(TestTableORM.a == 1).values(c='2')
>>> printf(update_stmt.compile(dialect=dialect, **COMPILE_KWARGS)) # 2
UPDATE db_table_orm SET c=%s WHERE db_table_orm.a = %s

0.7.9	-	UPDATE db_table_orm SET b=%s WHERE db_table_orm.a = %s
0.9.4	-	UPDATE db_table_orm SET b=%s WHERE db_table_orm.a = %s
1.0.11	-	UPDATE db_table_orm SET b=%s WHERE db_table_orm.a = %s
1.0.13	√	UPDATE db_table_orm SET c='2' WHERE db_table_orm.a = 1
1.1.6	√	UPDATE db_table_orm SET c='2' WHERE db_table_orm.a = 1

However, there is still a problem with unimplemented data types. If the parameters are substituted at all, an exception is raised:

>>> select_stmt = select([TestTable]).where(TestTable.c.cdate.in_((datetime.date(2016, 10, 3), datetime.date(2015, 8, 9)))).limit(1)
>>> printf(nts(select_stmt.compile(dialect=dialect, **COMPILE_KWARGS))) # 3 
Traceback (most recent call last):
...
NotImplementedError: Don't know how to literal-quote value datetime.date(2016, 10, 3)

0.7.9	-	No substitution of bind params.
0.9.4	-	NotImplementedError: Don't know how to literal-quote value datetime.date ...
1.0.11	-	NotImplementedError: Don't know how to literal-quote value datetime.date ...
1.0.13	-	NotImplementedError: Don't know how to literal-quote value datetime.date ...
1.1.6	-	NotImplementedError: Don't know how to literal-quote value datetime.date ...

Therefore sql_statements.LiteralDialect is still needed.

>>> dialect=LiteralDialect()

Unfortunately, it fails to compile INSERT statements with bind values of None, when using literal_binds=True for v1.0.13+.

>>> insert_stmt = insert(TestTable).values(**dict(((c, v) for c, v in zip((c_.name for c_ in TestTable.c), TEST_TABLE_DATA))))
>>> _compiled=insert_stmt.compile(dialect=dialect, **COMPILE_KWARGS)  # 4 

>>> printf(_compiled.__str__())  # 4 
INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext, cnull, cenum) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

0.7.9	-	INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
0.9.4	-	INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
1.0.11	-	INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
1.0.13	-	... CompileError: Bind parameter 'cnull' without a renderable value not allowed here.
1.1.6	-	... CompileError: Bind parameter 'cnull' without a renderable value not allowed here.

INSERT does work for v1.0.13+, if all None values for the INSERT statement are replaced with sqlalchemy.null instances:

>>> from sqlalchemy import null
>>> def null_if_none(value):
...     if value is None:
...         return null()
...     return value

>>> def fix_null_values(values):
...     return (null_if_none(v) for v in values)

>>> def fix_null_params(params):
...     if issequence(params):
...         # list of (col, val) tuples
...         return ((c, null_if_none(v)) for c, v in params)
...     return type(params)(((c, null_if_none(v)) for c, v in ditems(params)))

>>> insert_stmt = insert(TestTable).values(**dict(((c, v) for c, v in zip((c_.name for c_ in TestTable.c), fix_null_values(TEST_TABLE_DATA)))))
>>> _compiled = insert_stmt.compile(dialect=dialect, **COMPILE_KWARGS)

>>> printf(nts(_compiled.__str__()))  # 5 
INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext, cnull, cenum) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, NULL, ?)

This is just hilarious :)

Example usage:

fix_null_values(TEST_TABLE_DATA)
fix_null_params(zip((c_.name for c_ in TestTable.c), TEST_TABLE_DATA))
fix_null_params(dict(((c, v) for c, v in zip((c_.name for c_ in TestTable.c), TEST_TABLE_DATA))))

sql_statements_test.test_data_types()

Check correct rendering of various data types.

Tested with SLQAlchemy versions:

version	compile_kwargs
0.7.9	{}
0.9.4	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.11	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.13	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.6	{‘compile_kwargs’: {‘literal_binds’: True}}

See Test Environment for defintions.

>>> statement = select([TestTable]).where(TestTable.c.mycol.in_(TEST_TABLE_DATA)).limit(1)
>>> print(nts(sql_statement(statement))) 
SELECT db_table.mycol, db_table.cboolean, db_table.cinteger, db_table.cfloat, db_table.cdate, db_table.ctime, db_table.cdatetime, db_table.cstring, db_table.ctext, db_table.cnull, db_table.cenum
FROM db_table
WHERE db_table.mycol IN (5, ..., 100000000000000000000, 3.14159, '2016-10-03', '13:45:00', '2015-06-24 18:09:29.042517', 'UTF-8 snowman: ☃', 'snowman: ☃', NULL, 'foo')
 LIMIT 1...;

See python - SQLAlchemy: print the actual query - Stack Overflow

sql_statements_test.test_insert_update()

Check correct rendering of INSERT, UPDATE.

Tested with SLQAlchemy versions:

version	compile_kwargs
0.7.9	{}
0.9.4	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.11	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.13	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.6	{‘compile_kwargs’: {‘literal_binds’: True}}

See Test Environment for defintions.

>>> insert_stmt = insert(TestTableORM.__table__).values(a=1, b=2)
>>> print(sql_statement(insert_stmt))
INSERT INTO db_table_orm (b, a) VALUES (2, 1);

>>> update_stmt = update(TestTableORM.__table__).where(TestTableORM.a == 1).values(b=2)
>>> print(sql_statement(update_stmt))
UPDATE db_table_orm SET b=2 WHERE db_table_orm.a = 1;

See python - Sqlalchemy - how to get raw sql from insert(), update() statements with binded params? - Stack Overflow.

sql_statements_test.test_table_dump()

Check table dump convenience function.

Tested with SLQAlchemy versions:

version	compile_kwargs
0.7.9	{}
0.9.4	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.11	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.13	{‘compile_kwargs’: {‘literal_binds’: True}}
1.0.6	{‘compile_kwargs’: {‘literal_binds’: True}}

See Test Environment for defintions.

Check SQL table

>>> print(nts(table_dump_sql(TestTable, (TEST_TABLE_DATA, )))) 
CREATE TABLE db_table (
    mycol INTEGER NOT NULL...,
    cboolean BOOL...,
    cinteger INTEGER,
    cfloat FLOAT,
    cdate DATE,
    ctime TIME,
    cdatetime DATETIME,
    cstring VARCHAR(50),
    ctext TEXT,
    cnull INTEGER,
    cenum...,
    PRIMARY KEY (mycol),
    CHECK (cboolean IN (0, 1))...
);
INSERT INTO db_table (mycol, cboolean, cinteger, cfloat, cdate, ctime, cdatetime, cstring, ctext, cnull, cenum) VALUES (5, ..., 100000000000000000000, 3.14159, '2016-10-03', '13:45:00', '2015-06-24 18:09:29.042517', 'UTF-8 snowman: ☃', 'snowman: ☃', NULL, 'foo');

Check ORM table

>>> print(table_dump_sql(TestTableORM, [[1, 10 ** 20, '5'], [2, 6, 'text'], [3, 9, 'done']], 'a b c'.split())) 
CREATE TABLE db_table_orm (
    b INTEGER,
    c VARCHAR(10),
    a INTEGER NOT NULL...,
    PRIMARY KEY (a)
);
INSERT INTO db_table_orm (b, c, a) VALUES (100000000000000000000, '5', 1);
INSERT INTO db_table_orm (b, c, a) VALUES (6, 'text', 2);
INSERT INTO db_table_orm (b, c, a) VALUES (9, 'done', 3);

Check Update on ORM table

>>> print(table_dump_sql(TestTableORM, [[1, 10 ** 20, '5'], [2, 6, 'text'], [3, 9, 'done']], 'a b c'.split(), update=True, no_create=True))
UPDATE db_table_orm SET b=100000000000000000000, c='5', a=1 WHERE db_table_orm.a = 1;
UPDATE db_table_orm SET b=6, c='text', a=2 WHERE db_table_orm.a = 2;
UPDATE db_table_orm SET b=9, c='done', a=3 WHERE db_table_orm.a = 3;

Copyright

Copyright (C) 2017, Wolfgang Scherer, <wolfgang.scherer@gmx.de>. See the document source for conditions of use under the GNU Free Documentation License.