diff options
Diffstat (limited to 'test/dialect/mysql/test_compiler.py')
| -rw-r--r-- | test/dialect/mysql/test_compiler.py | 719 |
1 files changed, 410 insertions, 309 deletions
diff --git a/test/dialect/mysql/test_compiler.py b/test/dialect/mysql/test_compiler.py index 9e12e2d4c..79dff63a3 100644 --- a/test/dialect/mysql/test_compiler.py +++ b/test/dialect/mysql/test_compiler.py @@ -2,13 +2,48 @@ from sqlalchemy.testing import eq_, assert_raises_message, expect_warnings from sqlalchemy import sql, exc, schema, types as sqltypes -from sqlalchemy import Table, MetaData, Column, select, String, \ - Index, Integer, ForeignKey, PrimaryKeyConstraint, extract, \ - VARCHAR, NVARCHAR, Unicode, UnicodeText, \ - NUMERIC, DECIMAL, Numeric, Float, FLOAT, TIMESTAMP, DATE, \ - DATETIME, TIME, \ - DateTime, Time, Date, Interval, NCHAR, CHAR, CLOB, TEXT, Boolean, \ - BOOLEAN, LargeBinary, BLOB, SmallInteger, INT, func, cast, literal +from sqlalchemy import ( + Table, + MetaData, + Column, + select, + String, + Index, + Integer, + ForeignKey, + PrimaryKeyConstraint, + extract, + VARCHAR, + NVARCHAR, + Unicode, + UnicodeText, + NUMERIC, + DECIMAL, + Numeric, + Float, + FLOAT, + TIMESTAMP, + DATE, + DATETIME, + TIME, + DateTime, + Time, + Date, + Interval, + NCHAR, + CHAR, + CLOB, + TEXT, + Boolean, + BOOLEAN, + LargeBinary, + BLOB, + SmallInteger, + INT, + func, + cast, + literal, +) from sqlalchemy.dialects.mysql import insert from sqlalchemy.dialects.mysql import base as mysql @@ -22,237 +57,295 @@ class CompileTest(fixtures.TestBase, AssertsCompiledSQL): __dialect__ = mysql.dialect() def test_reserved_words(self): - table = Table("mysql_table", MetaData(), - Column("col1", Integer), - Column("master_ssl_verify_server_cert", Integer)) + table = Table( + "mysql_table", + MetaData(), + Column("col1", Integer), + Column("master_ssl_verify_server_cert", Integer), + ) x = select([table.c.col1, table.c.master_ssl_verify_server_cert]) self.assert_compile( x, "SELECT mysql_table.col1, " - "mysql_table.`master_ssl_verify_server_cert` FROM mysql_table") + "mysql_table.`master_ssl_verify_server_cert` FROM mysql_table", + ) def test_create_index_simple(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255))) - idx = Index('test_idx1', tbl.c.data) + tbl = Table("testtbl", m, Column("data", String(255))) + idx = Index("test_idx1", tbl.c.data) - self.assert_compile(schema.CreateIndex(idx), - 'CREATE INDEX test_idx1 ON testtbl (data)') + self.assert_compile( + schema.CreateIndex(idx), "CREATE INDEX test_idx1 ON testtbl (data)" + ) def test_create_index_with_prefix(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255))) - idx = Index('test_idx1', tbl.c.data, mysql_length=10, - mysql_prefix='FULLTEXT') + tbl = Table("testtbl", m, Column("data", String(255))) + idx = Index( + "test_idx1", tbl.c.data, mysql_length=10, mysql_prefix="FULLTEXT" + ) - self.assert_compile(schema.CreateIndex(idx), - 'CREATE FULLTEXT INDEX test_idx1 ' - 'ON testtbl (data(10))') + self.assert_compile( + schema.CreateIndex(idx), + "CREATE FULLTEXT INDEX test_idx1 " "ON testtbl (data(10))", + ) def test_create_index_with_parser(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255))) - idx = Index('test_idx1', tbl.c.data, mysql_length=10, - mysql_prefix='FULLTEXT', mysql_with_parser="ngram") + tbl = Table("testtbl", m, Column("data", String(255))) + idx = Index( + "test_idx1", + tbl.c.data, + mysql_length=10, + mysql_prefix="FULLTEXT", + mysql_with_parser="ngram", + ) - self.assert_compile(schema.CreateIndex(idx), - 'CREATE FULLTEXT INDEX test_idx1 ' - 'ON testtbl (data(10)) WITH PARSER ngram') + self.assert_compile( + schema.CreateIndex(idx), + "CREATE FULLTEXT INDEX test_idx1 " + "ON testtbl (data(10)) WITH PARSER ngram", + ) def test_create_index_with_length(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255))) - idx1 = Index('test_idx1', tbl.c.data, mysql_length=10) - idx2 = Index('test_idx2', tbl.c.data, mysql_length=5) + tbl = Table("testtbl", m, Column("data", String(255))) + idx1 = Index("test_idx1", tbl.c.data, mysql_length=10) + idx2 = Index("test_idx2", tbl.c.data, mysql_length=5) - self.assert_compile(schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl (data(10))') - self.assert_compile(schema.CreateIndex(idx2), - 'CREATE INDEX test_idx2 ON testtbl (data(5))') + self.assert_compile( + schema.CreateIndex(idx1), + "CREATE INDEX test_idx1 ON testtbl (data(10))", + ) + self.assert_compile( + schema.CreateIndex(idx2), + "CREATE INDEX test_idx2 ON testtbl (data(5))", + ) def test_create_index_with_length_quoted(self): m = MetaData() - tbl = Table('testtbl', m, Column('some quoted data', - String(255), key='s')) - idx1 = Index('test_idx1', tbl.c.s, mysql_length=10) + tbl = Table( + "testtbl", m, Column("some quoted data", String(255), key="s") + ) + idx1 = Index("test_idx1", tbl.c.s, mysql_length=10) self.assert_compile( schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl (`some quoted data`(10))') + "CREATE INDEX test_idx1 ON testtbl (`some quoted data`(10))", + ) def test_create_composite_index_with_length_quoted(self): m = MetaData() - tbl = Table('testtbl', m, - Column('some Quoted a', String(255), key='a'), - Column('some Quoted b', String(255), key='b')) - idx1 = Index('test_idx1', tbl.c.a, tbl.c.b, - mysql_length={'some Quoted a': 10, 'some Quoted b': 20}) + tbl = Table( + "testtbl", + m, + Column("some Quoted a", String(255), key="a"), + Column("some Quoted b", String(255), key="b"), + ) + idx1 = Index( + "test_idx1", + tbl.c.a, + tbl.c.b, + mysql_length={"some Quoted a": 10, "some Quoted b": 20}, + ) - self.assert_compile(schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl ' - '(`some Quoted a`(10), `some Quoted b`(20))') + self.assert_compile( + schema.CreateIndex(idx1), + "CREATE INDEX test_idx1 ON testtbl " + "(`some Quoted a`(10), `some Quoted b`(20))", + ) def test_create_composite_index_with_length_quoted_3085_workaround(self): m = MetaData() - tbl = Table('testtbl', m, - Column('some quoted a', String(255), key='a'), - Column('some quoted b', String(255), key='b')) + tbl = Table( + "testtbl", + m, + Column("some quoted a", String(255), key="a"), + Column("some quoted b", String(255), key="b"), + ) idx1 = Index( - 'test_idx1', tbl.c.a, tbl.c.b, - mysql_length={'`some quoted a`': 10, '`some quoted b`': 20} + "test_idx1", + tbl.c.a, + tbl.c.b, + mysql_length={"`some quoted a`": 10, "`some quoted b`": 20}, ) - self.assert_compile(schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl ' - '(`some quoted a`(10), `some quoted b`(20))') + self.assert_compile( + schema.CreateIndex(idx1), + "CREATE INDEX test_idx1 ON testtbl " + "(`some quoted a`(10), `some quoted b`(20))", + ) def test_create_composite_index_with_length(self): m = MetaData() - tbl = Table('testtbl', m, - Column('a', String(255)), - Column('b', String(255))) + tbl = Table( + "testtbl", m, Column("a", String(255)), Column("b", String(255)) + ) - idx1 = Index('test_idx1', tbl.c.a, tbl.c.b, - mysql_length={'a': 10, 'b': 20}) - idx2 = Index('test_idx2', tbl.c.a, tbl.c.b, - mysql_length={'a': 15}) - idx3 = Index('test_idx3', tbl.c.a, tbl.c.b, - mysql_length=30) + idx1 = Index( + "test_idx1", tbl.c.a, tbl.c.b, mysql_length={"a": 10, "b": 20} + ) + idx2 = Index("test_idx2", tbl.c.a, tbl.c.b, mysql_length={"a": 15}) + idx3 = Index("test_idx3", tbl.c.a, tbl.c.b, mysql_length=30) self.assert_compile( schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl (a(10), b(20))' + "CREATE INDEX test_idx1 ON testtbl (a(10), b(20))", ) self.assert_compile( schema.CreateIndex(idx2), - 'CREATE INDEX test_idx2 ON testtbl (a(15), b)' + "CREATE INDEX test_idx2 ON testtbl (a(15), b)", ) self.assert_compile( schema.CreateIndex(idx3), - 'CREATE INDEX test_idx3 ON testtbl (a(30), b(30))' + "CREATE INDEX test_idx3 ON testtbl (a(30), b(30))", ) def test_create_index_with_using(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255))) - idx1 = Index('test_idx1', tbl.c.data, mysql_using='btree') - idx2 = Index('test_idx2', tbl.c.data, mysql_using='hash') + tbl = Table("testtbl", m, Column("data", String(255))) + idx1 = Index("test_idx1", tbl.c.data, mysql_using="btree") + idx2 = Index("test_idx2", tbl.c.data, mysql_using="hash") self.assert_compile( schema.CreateIndex(idx1), - 'CREATE INDEX test_idx1 ON testtbl (data) USING btree') + "CREATE INDEX test_idx1 ON testtbl (data) USING btree", + ) self.assert_compile( schema.CreateIndex(idx2), - 'CREATE INDEX test_idx2 ON testtbl (data) USING hash') + "CREATE INDEX test_idx2 ON testtbl (data) USING hash", + ) def test_create_pk_plain(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255)), - PrimaryKeyConstraint('data')) + tbl = Table( + "testtbl", + m, + Column("data", String(255)), + PrimaryKeyConstraint("data"), + ) self.assert_compile( schema.CreateTable(tbl), "CREATE TABLE testtbl (data VARCHAR(255) NOT NULL, " - "PRIMARY KEY (data))") + "PRIMARY KEY (data))", + ) def test_create_pk_with_using(self): m = MetaData() - tbl = Table('testtbl', m, Column('data', String(255)), - PrimaryKeyConstraint('data', mysql_using='btree')) + tbl = Table( + "testtbl", + m, + Column("data", String(255)), + PrimaryKeyConstraint("data", mysql_using="btree"), + ) self.assert_compile( schema.CreateTable(tbl), "CREATE TABLE testtbl (data VARCHAR(255) NOT NULL, " - "PRIMARY KEY (data) USING btree)") + "PRIMARY KEY (data) USING btree)", + ) def test_create_index_expr(self): m = MetaData() - t1 = Table('foo', m, - Column('x', Integer) - ) + t1 = Table("foo", m, Column("x", Integer)) self.assert_compile( schema.CreateIndex(Index("bar", t1.c.x > 5)), - "CREATE INDEX bar ON foo (x > 5)" + "CREATE INDEX bar ON foo (x > 5)", ) def test_deferrable_initially_kw_not_ignored(self): m = MetaData() - Table('t1', m, Column('id', Integer, primary_key=True)) + Table("t1", m, Column("id", Integer, primary_key=True)) t2 = Table( - 't2', m, Column( - 'id', Integer, - ForeignKey('t1.id', deferrable=True, initially="XYZ"), - primary_key=True)) + "t2", + m, + Column( + "id", + Integer, + ForeignKey("t1.id", deferrable=True, initially="XYZ"), + primary_key=True, + ), + ) self.assert_compile( schema.CreateTable(t2), "CREATE TABLE t2 (id INTEGER NOT NULL, " "PRIMARY KEY (id), FOREIGN KEY(id) REFERENCES t1 (id) " - "DEFERRABLE INITIALLY XYZ)" + "DEFERRABLE INITIALLY XYZ)", ) def test_match_kw_raises(self): m = MetaData() - Table('t1', m, Column('id', Integer, primary_key=True)) - t2 = Table('t2', m, Column('id', Integer, - ForeignKey('t1.id', match="XYZ"), - primary_key=True)) + Table("t1", m, Column("id", Integer, primary_key=True)) + t2 = Table( + "t2", + m, + Column( + "id", + Integer, + ForeignKey("t1.id", match="XYZ"), + primary_key=True, + ), + ) assert_raises_message( exc.CompileError, "MySQL ignores the 'MATCH' keyword while at the same time causes " "ON UPDATE/ON DELETE clauses to be ignored.", - schema.CreateTable(t2).compile, dialect=mysql.dialect() + schema.CreateTable(t2).compile, + dialect=mysql.dialect(), ) def test_match(self): - matchtable = table('matchtable', column('title', String)) + matchtable = table("matchtable", column("title", String)) self.assert_compile( - matchtable.c.title.match('somstr'), - "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)") + matchtable.c.title.match("somstr"), + "MATCH (matchtable.title) AGAINST (%s IN BOOLEAN MODE)", + ) def test_match_compile_kw(self): - expr = literal('x').match(literal('y')) + expr = literal("x").match(literal("y")) self.assert_compile( expr, "MATCH ('x') AGAINST ('y' IN BOOLEAN MODE)", - literal_binds=True + literal_binds=True, ) def test_concat_compile_kw(self): - expr = literal('x', type_=String) + literal('y', type_=String) - self.assert_compile( - expr, - "concat('x', 'y')", - literal_binds=True - ) + expr = literal("x", type_=String) + literal("y", type_=String) + self.assert_compile(expr, "concat('x', 'y')", literal_binds=True) def test_for_update(self): - table1 = table('mytable', - column('myid'), column('name'), column('description')) + table1 = table( + "mytable", column("myid"), column("name"), column("description") + ) self.assert_compile( table1.select(table1.c.myid == 7).with_for_update(), "SELECT mytable.myid, mytable.name, mytable.description " - "FROM mytable WHERE mytable.myid = %s FOR UPDATE") + "FROM mytable WHERE mytable.myid = %s FOR UPDATE", + ) self.assert_compile( table1.select(table1.c.myid == 7).with_for_update(read=True), "SELECT mytable.myid, mytable.name, mytable.description " - "FROM mytable WHERE mytable.myid = %s LOCK IN SHARE MODE") + "FROM mytable WHERE mytable.myid = %s LOCK IN SHARE MODE", + ) def test_delete_extra_froms(self): - t1 = table('t1', column('c1')) - t2 = table('t2', column('c1')) + t1 = table("t1", column("c1")) + t2 = table("t2", column("c1")) q = sql.delete(t1).where(t1.c.c1 == t2.c.c1) self.assert_compile( q, "DELETE FROM t1 USING t1, t2 WHERE t1.c1 = t2.c1" ) def test_delete_extra_froms_alias(self): - a1 = table('t1', column('c1')).alias('a1') - t2 = table('t2', column('c1')) + a1 = table("t1", column("c1")).alias("a1") + t2 = table("t2", column("c1")) q = sql.delete(a1).where(a1.c.c1 == t2.c.c1) self.assert_compile( q, "DELETE FROM a1 USING t1 AS a1, t2 WHERE a1.c1 = t2.c1" @@ -272,60 +365,60 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): def gen(distinct=None, prefixes=None): kw = {} if distinct is not None: - kw['distinct'] = distinct + kw["distinct"] = distinct if prefixes is not None: - kw['prefixes'] = prefixes - return str(select([column('q')], **kw).compile(dialect=dialect)) + kw["prefixes"] = prefixes + return str(select([column("q")], **kw).compile(dialect=dialect)) - eq_(gen(None), 'SELECT q') - eq_(gen(True), 'SELECT DISTINCT q') + eq_(gen(None), "SELECT q") + eq_(gen(True), "SELECT DISTINCT q") - eq_(gen(prefixes=['ALL']), 'SELECT ALL q') - eq_(gen(prefixes=['DISTINCTROW']), - 'SELECT DISTINCTROW q') + eq_(gen(prefixes=["ALL"]), "SELECT ALL q") + eq_(gen(prefixes=["DISTINCTROW"]), "SELECT DISTINCTROW q") # Interaction with MySQL prefix extensions + eq_(gen(None, ["straight_join"]), "SELECT straight_join q") eq_( - gen(None, ['straight_join']), - 'SELECT straight_join q') - eq_( - gen(False, ['HIGH_PRIORITY', 'SQL_SMALL_RESULT', 'ALL']), - 'SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q') + gen(False, ["HIGH_PRIORITY", "SQL_SMALL_RESULT", "ALL"]), + "SELECT HIGH_PRIORITY SQL_SMALL_RESULT ALL q", + ) eq_( - gen(True, ['high_priority', sql.text('sql_cache')]), - 'SELECT high_priority sql_cache DISTINCT q') + gen(True, ["high_priority", sql.text("sql_cache")]), + "SELECT high_priority sql_cache DISTINCT q", + ) def test_backslash_escaping(self): self.assert_compile( - sql.column('foo').like('bar', escape='\\'), - "foo LIKE %s ESCAPE '\\\\'" + sql.column("foo").like("bar", escape="\\"), + "foo LIKE %s ESCAPE '\\\\'", ) dialect = mysql.dialect() dialect._backslash_escapes = False self.assert_compile( - sql.column('foo').like('bar', escape='\\'), + sql.column("foo").like("bar", escape="\\"), "foo LIKE %s ESCAPE '\\'", - dialect=dialect + dialect=dialect, ) def test_limit(self): - t = sql.table('t', sql.column('col1'), sql.column('col2')) + t = sql.table("t", sql.column("col1"), sql.column("col2")) self.assert_compile( select([t]).limit(10).offset(20), "SELECT t.col1, t.col2 FROM t LIMIT %s, %s", - {'param_1': 20, 'param_2': 10} + {"param_1": 20, "param_2": 10}, ) self.assert_compile( select([t]).limit(10), "SELECT t.col1, t.col2 FROM t LIMIT %s", - {'param_1': 10}) + {"param_1": 10}, + ) self.assert_compile( select([t]).offset(10), "SELECT t.col1, t.col2 FROM t LIMIT %s, 18446744073709551615", - {'param_1': 10} + {"param_1": 10}, ) def test_varchar_raise(self): @@ -343,38 +436,35 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): exc.CompileError, "VARCHAR requires a length on dialect mysql", type_.compile, - dialect=mysql.dialect() + dialect=mysql.dialect(), ) - t1 = Table('sometable', MetaData(), - Column('somecolumn', type_) - ) + t1 = Table("sometable", MetaData(), Column("somecolumn", type_)) assert_raises_message( exc.CompileError, r"\(in table 'sometable', column 'somecolumn'\)\: " r"(?:N)?VARCHAR requires a length on dialect mysql", schema.CreateTable(t1).compile, - dialect=mysql.dialect() + dialect=mysql.dialect(), ) def test_update_limit(self): - t = sql.table('t', sql.column('col1'), sql.column('col2')) + t = sql.table("t", sql.column("col1"), sql.column("col2")) self.assert_compile( - t.update(values={'col1': 123}), - "UPDATE t SET col1=%s" + t.update(values={"col1": 123}), "UPDATE t SET col1=%s" ) self.assert_compile( - t.update(values={'col1': 123}, mysql_limit=5), - "UPDATE t SET col1=%s LIMIT 5" + t.update(values={"col1": 123}, mysql_limit=5), + "UPDATE t SET col1=%s LIMIT 5", ) self.assert_compile( - t.update(values={'col1': 123}, mysql_limit=None), - "UPDATE t SET col1=%s" + t.update(values={"col1": 123}, mysql_limit=None), + "UPDATE t SET col1=%s", ) self.assert_compile( - t.update(t.c.col2 == 456, values={'col1': 123}, mysql_limit=1), - "UPDATE t SET col1=%s WHERE t.col2 = %s LIMIT 1" + t.update(t.c.col2 == 456, values={"col1": 123}, mysql_limit=1), + "UPDATE t SET col1=%s WHERE t.col2 = %s LIMIT 1", ) def test_utc_timestamp(self): @@ -382,14 +472,16 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): def test_utc_timestamp_fsp(self): self.assert_compile( - func.utc_timestamp(5), "utc_timestamp(%s)", - checkparams={"utc_timestamp_1": 5}) + func.utc_timestamp(5), + "utc_timestamp(%s)", + checkparams={"utc_timestamp_1": 5}, + ) def test_sysdate(self): self.assert_compile(func.sysdate(), "SYSDATE()") def test_cast(self): - t = sql.table('t', sql.column('col')) + t = sql.table("t", sql.column("col")) m = mysql specs = [ @@ -403,16 +495,13 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): # 'SIGNED INTEGER' is a bigint, so this is ok. (m.MSBigInteger, "CAST(t.col AS SIGNED INTEGER)"), (m.MSBigInteger(unsigned=False), "CAST(t.col AS SIGNED INTEGER)"), - (m.MSBigInteger(unsigned=True), - "CAST(t.col AS UNSIGNED INTEGER)"), - + (m.MSBigInteger(unsigned=True), "CAST(t.col AS UNSIGNED INTEGER)"), # this is kind of sucky. thank you default arguments! (NUMERIC, "CAST(t.col AS DECIMAL)"), (DECIMAL, "CAST(t.col AS DECIMAL)"), (Numeric, "CAST(t.col AS DECIMAL)"), (m.MSNumeric, "CAST(t.col AS DECIMAL)"), (m.MSDecimal, "CAST(t.col AS DECIMAL)"), - (TIMESTAMP, "CAST(t.col AS DATETIME)"), (DATETIME, "CAST(t.col AS DATETIME)"), (DATE, "CAST(t.col AS DATE)"), @@ -424,17 +513,16 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): (Date, "CAST(t.col AS DATE)"), (m.MSTime, "CAST(t.col AS TIME)"), (m.MSTimeStamp, "CAST(t.col AS DATETIME)"), - (String, "CAST(t.col AS CHAR)"), (Unicode, "CAST(t.col AS CHAR)"), (UnicodeText, "CAST(t.col AS CHAR)"), (VARCHAR, "CAST(t.col AS CHAR)"), (NCHAR, "CAST(t.col AS CHAR)"), (CHAR, "CAST(t.col AS CHAR)"), - (m.CHAR(charset='utf8'), "CAST(t.col AS CHAR CHARACTER SET utf8)"), + (m.CHAR(charset="utf8"), "CAST(t.col AS CHAR CHARACTER SET utf8)"), (CLOB, "CAST(t.col AS CHAR)"), (TEXT, "CAST(t.col AS CHAR)"), - (m.TEXT(charset='utf8'), "CAST(t.col AS CHAR CHARACTER SET utf8)"), + (m.TEXT(charset="utf8"), "CAST(t.col AS CHAR CHARACTER SET utf8)"), (String(32), "CAST(t.col AS CHAR(32))"), (Unicode(32), "CAST(t.col AS CHAR(32))"), (CHAR(32), "CAST(t.col AS CHAR(32))"), @@ -445,7 +533,6 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): (m.MSLongText, "CAST(t.col AS CHAR)"), (m.MSNChar, "CAST(t.col AS CHAR)"), (m.MSNVarChar, "CAST(t.col AS CHAR)"), - (LargeBinary, "CAST(t.col AS BINARY)"), (BLOB, "CAST(t.col AS BINARY)"), (m.MSBlob, "CAST(t.col AS BINARY)"), @@ -457,9 +544,7 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): (m.MSBinary(32), "CAST(t.col AS BINARY)"), (m.MSVarBinary, "CAST(t.col AS BINARY)"), (m.MSVarBinary(32), "CAST(t.col AS BINARY)"), - (Interval, "CAST(t.col AS DATETIME)"), - ] for type_, expected in specs: @@ -470,63 +555,52 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): impl = Integer type_ = MyInteger() - t = sql.table('t', sql.column('col')) + t = sql.table("t", sql.column("col")) self.assert_compile( - cast(t.c.col, type_), "CAST(t.col AS SIGNED INTEGER)") + cast(t.c.col, type_), "CAST(t.col AS SIGNED INTEGER)" + ) def test_cast_literal_bind(self): - expr = cast(column('foo', Integer) + 5, Integer()) + expr = cast(column("foo", Integer) + 5, Integer()) self.assert_compile( - expr, - "CAST(foo + 5 AS SIGNED INTEGER)", - literal_binds=True + expr, "CAST(foo + 5 AS SIGNED INTEGER)", literal_binds=True ) def test_unsupported_cast_literal_bind(self): - expr = cast(column('foo', Integer) + 5, Float) + expr = cast(column("foo", Integer) + 5, Float) - with expect_warnings( - "Datatype FLOAT does not support CAST on MySQL;" - ): - self.assert_compile( - expr, - "(foo + 5)", - literal_binds=True - ) + with expect_warnings("Datatype FLOAT does not support CAST on MySQL;"): + self.assert_compile(expr, "(foo + 5)", literal_binds=True) dialect = mysql.MySQLDialect() dialect.server_version_info = (3, 9, 8) - with expect_warnings( - "Current MySQL version does not support CAST" - ): + with expect_warnings("Current MySQL version does not support CAST"): eq_( - str(expr.compile( - dialect=dialect, - compile_kwargs={"literal_binds": True})), - "(foo + 5)" + str( + expr.compile( + dialect=dialect, compile_kwargs={"literal_binds": True} + ) + ), + "(foo + 5)", ) def test_unsupported_casts(self): - t = sql.table('t', sql.column('col')) + t = sql.table("t", sql.column("col")) m = mysql specs = [ (m.MSBit, "t.col"), - (FLOAT, "t.col"), (Float, "t.col"), (m.MSFloat, "t.col"), (m.MSDouble, "t.col"), (m.MSReal, "t.col"), - (m.MSYear, "t.col"), (m.MSYear(2), "t.col"), - (Boolean, "t.col"), (BOOLEAN, "t.col"), - (m.MSEnum, "t.col"), (m.MSEnum("1", "2"), "t.col"), (m.MSSet, "t.col"), @@ -541,23 +615,19 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): def test_no_cast_pre_4(self): self.assert_compile( - cast(Column('foo', Integer), String), - "CAST(foo AS CHAR)", + cast(Column("foo", Integer), String), "CAST(foo AS CHAR)" ) dialect = mysql.dialect() dialect.server_version_info = (3, 2, 3) with expect_warnings("Current MySQL version does not support CAST;"): self.assert_compile( - cast(Column('foo', Integer), String), - "foo", - dialect=dialect + cast(Column("foo", Integer), String), "foo", dialect=dialect ) def test_cast_grouped_expression_non_castable(self): with expect_warnings("Datatype FLOAT does not support CAST on MySQL;"): self.assert_compile( - cast(sql.column('x') + sql.column('y'), Float), - "(x + y)" + cast(sql.column("x") + sql.column("y"), Float), "(x + y)" ) def test_cast_grouped_expression_pre_4(self): @@ -565,173 +635,199 @@ class SQLTest(fixtures.TestBase, AssertsCompiledSQL): dialect.server_version_info = (3, 2, 3) with expect_warnings("Current MySQL version does not support CAST;"): self.assert_compile( - cast(sql.column('x') + sql.column('y'), Integer), + cast(sql.column("x") + sql.column("y"), Integer), "(x + y)", - dialect=dialect + dialect=dialect, ) def test_extract(self): - t = sql.table('t', sql.column('col1')) + t = sql.table("t", sql.column("col1")) - for field in 'year', 'month', 'day': + for field in "year", "month", "day": self.assert_compile( select([extract(field, t.c.col1)]), - "SELECT EXTRACT(%s FROM t.col1) AS anon_1 FROM t" % field) + "SELECT EXTRACT(%s FROM t.col1) AS anon_1 FROM t" % field, + ) # millsecondS to millisecond self.assert_compile( - select([extract('milliseconds', t.c.col1)]), - "SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t") + select([extract("milliseconds", t.c.col1)]), + "SELECT EXTRACT(millisecond FROM t.col1) AS anon_1 FROM t", + ) def test_too_long_index(self): - exp = 'ix_zyrenian_zyme_zyzzogeton_zyzzogeton_zyrenian_zyme_zyz_5cd2' - tname = 'zyrenian_zyme_zyzzogeton_zyzzogeton' - cname = 'zyrenian_zyme_zyzzogeton_zo' + exp = "ix_zyrenian_zyme_zyzzogeton_zyzzogeton_zyrenian_zyme_zyz_5cd2" + tname = "zyrenian_zyme_zyzzogeton_zyzzogeton" + cname = "zyrenian_zyme_zyzzogeton_zo" - t1 = Table(tname, MetaData(), - Column(cname, Integer, index=True), - ) + t1 = Table(tname, MetaData(), Column(cname, Integer, index=True)) ix1 = list(t1.indexes)[0] self.assert_compile( schema.CreateIndex(ix1), - "CREATE INDEX %s " - "ON %s (%s)" % (exp, tname, cname) + "CREATE INDEX %s " "ON %s (%s)" % (exp, tname, cname), ) def test_innodb_autoincrement(self): t1 = Table( - 'sometable', MetaData(), + "sometable", + MetaData(), Column( - 'assigned_id', Integer(), primary_key=True, - autoincrement=False), - Column('id', Integer(), primary_key=True, autoincrement=True), - mysql_engine='InnoDB') - self.assert_compile(schema.CreateTable(t1), - 'CREATE TABLE sometable (assigned_id ' - 'INTEGER NOT NULL, id INTEGER NOT NULL ' - 'AUTO_INCREMENT, PRIMARY KEY (id, assigned_id)' - ')ENGINE=InnoDB') - - t1 = Table('sometable', MetaData(), - Column('assigned_id', Integer(), primary_key=True, - autoincrement=True), - Column('id', Integer(), primary_key=True, - autoincrement=False), mysql_engine='InnoDB') - self.assert_compile(schema.CreateTable(t1), - 'CREATE TABLE sometable (assigned_id ' - 'INTEGER NOT NULL AUTO_INCREMENT, id ' - 'INTEGER NOT NULL, PRIMARY KEY ' - '(assigned_id, id))ENGINE=InnoDB') + "assigned_id", Integer(), primary_key=True, autoincrement=False + ), + Column("id", Integer(), primary_key=True, autoincrement=True), + mysql_engine="InnoDB", + ) + self.assert_compile( + schema.CreateTable(t1), + "CREATE TABLE sometable (assigned_id " + "INTEGER NOT NULL, id INTEGER NOT NULL " + "AUTO_INCREMENT, PRIMARY KEY (id, assigned_id)" + ")ENGINE=InnoDB", + ) + + t1 = Table( + "sometable", + MetaData(), + Column( + "assigned_id", Integer(), primary_key=True, autoincrement=True + ), + Column("id", Integer(), primary_key=True, autoincrement=False), + mysql_engine="InnoDB", + ) + self.assert_compile( + schema.CreateTable(t1), + "CREATE TABLE sometable (assigned_id " + "INTEGER NOT NULL AUTO_INCREMENT, id " + "INTEGER NOT NULL, PRIMARY KEY " + "(assigned_id, id))ENGINE=InnoDB", + ) def test_innodb_autoincrement_reserved_word_column_name(self): t1 = Table( - 'sometable', MetaData(), - Column('id', Integer(), primary_key=True, autoincrement=False), - Column('order', Integer(), primary_key=True, autoincrement=True), - mysql_engine='InnoDB') + "sometable", + MetaData(), + Column("id", Integer(), primary_key=True, autoincrement=False), + Column("order", Integer(), primary_key=True, autoincrement=True), + mysql_engine="InnoDB", + ) self.assert_compile( schema.CreateTable(t1), - 'CREATE TABLE sometable (' - 'id INTEGER NOT NULL, ' - '`order` INTEGER NOT NULL AUTO_INCREMENT, ' - 'PRIMARY KEY (`order`, id)' - ')ENGINE=InnoDB') + "CREATE TABLE sometable (" + "id INTEGER NOT NULL, " + "`order` INTEGER NOT NULL AUTO_INCREMENT, " + "PRIMARY KEY (`order`, id)" + ")ENGINE=InnoDB", + ) def test_create_table_with_partition(self): t1 = Table( - 'testtable', MetaData(), - Column('id', Integer(), primary_key=True, autoincrement=True), - Column('other_id', Integer(), primary_key=True, - autoincrement=False), - mysql_partitions='2', mysql_partition_by='KEY(other_id)') + "testtable", + MetaData(), + Column("id", Integer(), primary_key=True, autoincrement=True), + Column( + "other_id", Integer(), primary_key=True, autoincrement=False + ), + mysql_partitions="2", + mysql_partition_by="KEY(other_id)", + ) self.assert_compile( schema.CreateTable(t1), - 'CREATE TABLE testtable (' - 'id INTEGER NOT NULL AUTO_INCREMENT, ' - 'other_id INTEGER NOT NULL, ' - 'PRIMARY KEY (id, other_id)' - ')PARTITION BY KEY(other_id) PARTITIONS 2' + "CREATE TABLE testtable (" + "id INTEGER NOT NULL AUTO_INCREMENT, " + "other_id INTEGER NOT NULL, " + "PRIMARY KEY (id, other_id)" + ")PARTITION BY KEY(other_id) PARTITIONS 2", ) def test_create_table_with_subpartition(self): t1 = Table( - 'testtable', MetaData(), - Column('id', Integer(), primary_key=True, autoincrement=True), - Column('other_id', Integer(), primary_key=True, - autoincrement=False), - mysql_partitions='2', - mysql_partition_by='KEY(other_id)', + "testtable", + MetaData(), + Column("id", Integer(), primary_key=True, autoincrement=True), + Column( + "other_id", Integer(), primary_key=True, autoincrement=False + ), + mysql_partitions="2", + mysql_partition_by="KEY(other_id)", mysql_subpartition_by="HASH(some_expr)", - mysql_subpartitions='2') + mysql_subpartitions="2", + ) self.assert_compile( schema.CreateTable(t1), - 'CREATE TABLE testtable (' - 'id INTEGER NOT NULL AUTO_INCREMENT, ' - 'other_id INTEGER NOT NULL, ' - 'PRIMARY KEY (id, other_id)' - ')PARTITION BY KEY(other_id) PARTITIONS 2 ' - 'SUBPARTITION BY HASH(some_expr) SUBPARTITIONS 2' + "CREATE TABLE testtable (" + "id INTEGER NOT NULL AUTO_INCREMENT, " + "other_id INTEGER NOT NULL, " + "PRIMARY KEY (id, other_id)" + ")PARTITION BY KEY(other_id) PARTITIONS 2 " + "SUBPARTITION BY HASH(some_expr) SUBPARTITIONS 2", ) def test_create_table_with_partition_hash(self): t1 = Table( - 'testtable', MetaData(), - Column('id', Integer(), primary_key=True, autoincrement=True), - Column('other_id', Integer(), primary_key=True, - autoincrement=False), - mysql_partitions='2', mysql_partition_by='HASH(other_id)') + "testtable", + MetaData(), + Column("id", Integer(), primary_key=True, autoincrement=True), + Column( + "other_id", Integer(), primary_key=True, autoincrement=False + ), + mysql_partitions="2", + mysql_partition_by="HASH(other_id)", + ) self.assert_compile( schema.CreateTable(t1), - 'CREATE TABLE testtable (' - 'id INTEGER NOT NULL AUTO_INCREMENT, ' - 'other_id INTEGER NOT NULL, ' - 'PRIMARY KEY (id, other_id)' - ')PARTITION BY HASH(other_id) PARTITIONS 2' + "CREATE TABLE testtable (" + "id INTEGER NOT NULL AUTO_INCREMENT, " + "other_id INTEGER NOT NULL, " + "PRIMARY KEY (id, other_id)" + ")PARTITION BY HASH(other_id) PARTITIONS 2", ) def test_create_table_with_partition_and_other_opts(self): t1 = Table( - 'testtable', MetaData(), - Column('id', Integer(), primary_key=True, autoincrement=True), - Column('other_id', Integer(), primary_key=True, - autoincrement=False), - mysql_stats_sample_pages='2', - mysql_partitions='2', mysql_partition_by='HASH(other_id)') + "testtable", + MetaData(), + Column("id", Integer(), primary_key=True, autoincrement=True), + Column( + "other_id", Integer(), primary_key=True, autoincrement=False + ), + mysql_stats_sample_pages="2", + mysql_partitions="2", + mysql_partition_by="HASH(other_id)", + ) self.assert_compile( schema.CreateTable(t1), - 'CREATE TABLE testtable (' - 'id INTEGER NOT NULL AUTO_INCREMENT, ' - 'other_id INTEGER NOT NULL, ' - 'PRIMARY KEY (id, other_id)' - ')STATS_SAMPLE_PAGES=2 PARTITION BY HASH(other_id) PARTITIONS 2' + "CREATE TABLE testtable (" + "id INTEGER NOT NULL AUTO_INCREMENT, " + "other_id INTEGER NOT NULL, " + "PRIMARY KEY (id, other_id)" + ")STATS_SAMPLE_PAGES=2 PARTITION BY HASH(other_id) PARTITIONS 2", ) def test_inner_join(self): - t1 = table('t1', column('x')) - t2 = table('t2', column('y')) + t1 = table("t1", column("x")) + t2 = table("t2", column("y")) self.assert_compile( - t1.join(t2, t1.c.x == t2.c.y), - "t1 INNER JOIN t2 ON t1.x = t2.y" + t1.join(t2, t1.c.x == t2.c.y), "t1 INNER JOIN t2 ON t1.x = t2.y" ) def test_outer_join(self): - t1 = table('t1', column('x')) - t2 = table('t2', column('y')) + t1 = table("t1", column("x")) + t2 = table("t2", column("y")) self.assert_compile( t1.outerjoin(t2, t1.c.x == t2.c.y), - "t1 LEFT OUTER JOIN t2 ON t1.x = t2.y" + "t1 LEFT OUTER JOIN t2 ON t1.x = t2.y", ) def test_full_outer_join(self): - t1 = table('t1', column('x')) - t2 = table('t2', column('y')) + t1 = table("t1", column("x")) + t2 = table("t2", column("y")) self.assert_compile( t1.outerjoin(t2, t1.c.x == t2.c.y, full=True), - "t1 FULL OUTER JOIN t2 ON t1.x = t2.y" + "t1 FULL OUTER JOIN t2 ON t1.x = t2.y", ) @@ -740,39 +836,44 @@ class InsertOnDuplicateTest(fixtures.TestBase, AssertsCompiledSQL): def setup(self): self.table = Table( - 'foos', MetaData(), - Column('id', Integer, primary_key=True), - Column('bar', String(10)), - Column('baz', String(10)), + "foos", + MetaData(), + Column("id", Integer, primary_key=True), + Column("bar", String(10)), + Column("baz", String(10)), ) def test_from_values(self): stmt = insert(self.table).values( - [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}]) + [{"id": 1, "bar": "ab"}, {"id": 2, "bar": "b"}] + ) stmt = stmt.on_duplicate_key_update( - bar=stmt.inserted.bar, baz=stmt.inserted.baz) + bar=stmt.inserted.bar, baz=stmt.inserted.baz + ) expected_sql = ( - 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) ' - 'ON DUPLICATE KEY UPDATE bar = VALUES(bar), baz = VALUES(baz)' + "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) " + "ON DUPLICATE KEY UPDATE bar = VALUES(bar), baz = VALUES(baz)" ) self.assert_compile(stmt, expected_sql) def test_from_literal(self): stmt = insert(self.table).values( - [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}]) - stmt = stmt.on_duplicate_key_update(bar=literal_column('bb')) + [{"id": 1, "bar": "ab"}, {"id": 2, "bar": "b"}] + ) + stmt = stmt.on_duplicate_key_update(bar=literal_column("bb")) expected_sql = ( - 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) ' - 'ON DUPLICATE KEY UPDATE bar = bb' + "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) " + "ON DUPLICATE KEY UPDATE bar = bb" ) self.assert_compile(stmt, expected_sql) def test_python_values(self): stmt = insert(self.table).values( - [{'id': 1, 'bar': 'ab'}, {'id': 2, 'bar': 'b'}]) + [{"id": 1, "bar": "ab"}, {"id": 2, "bar": "b"}] + ) stmt = stmt.on_duplicate_key_update(bar="foobar") expected_sql = ( - 'INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) ' - 'ON DUPLICATE KEY UPDATE bar = %s' + "INSERT INTO foos (id, bar) VALUES (%s, %s), (%s, %s) " + "ON DUPLICATE KEY UPDATE bar = %s" ) self.assert_compile(stmt, expected_sql) |