pets = relationship(
"Pets",
primaryjoin=(
- "foreign(Pets.person_id)" "==cast(type_coerce(Person.id, Integer), Integer)"
+ "foreign(Pets.person_id)==cast(type_coerce(Person.id, Integer), Integer)"
),
)
name = mapped_column(String)
boston_addresses = relationship(
"Address",
- primaryjoin="and_(User.id==Address.user_id, " "Address.city=='Boston')",
+ primaryjoin="and_(User.id==Address.user_id, Address.city=='Boston')",
)
network = relationship(
"Network",
- primaryjoin="IPA.v4address.bool_op('<<')" "(foreign(Network.v4representation))",
+ primaryjoin="IPA.v4address.bool_op('<<')(foreign(Network.v4representation))",
viewonly=True,
)
d = relationship(
"D",
- secondary="join(B, D, B.d_id == D.id)." "join(C, C.d_id == D.id)",
+ secondary="join(B, D, B.d_id == D.id).join(C, C.d_id == D.id)",
primaryjoin="and_(A.b_id == B.id, A.id == C.a_id)",
secondaryjoin="D.id == B.d_id",
uselist=False,
)
NO_ARG_FNS = set(
- "UID CURRENT_DATE SYSDATE USER " "CURRENT_TIME CURRENT_TIMESTAMP".split()
+ "UID CURRENT_DATE SYSDATE USER CURRENT_TIME CURRENT_TIMESTAMP".split()
)
"""
if not self.context.compiled:
raise exc.InvalidRequestError(
- "Statement is not a compiled " "expression construct."
+ "Statement is not a compiled expression construct."
)
elif not self.context.isinsert:
raise exc.InvalidRequestError(
- "Statement is not an insert() " "expression construct."
+ "Statement is not an insert() expression construct."
)
elif self.context._is_explicit_returning:
raise exc.InvalidRequestError(
"""
if not self.context.compiled:
raise exc.InvalidRequestError(
- "Statement is not a compiled " "expression construct."
+ "Statement is not a compiled expression construct."
)
elif not self.context.isupdate:
raise exc.InvalidRequestError(
- "Statement is not an update() " "expression construct."
+ "Statement is not an update() expression construct."
)
elif self.context.executemany:
return self.context.compiled_parameters
"""
if not self.context.compiled:
raise exc.InvalidRequestError(
- "Statement is not a compiled " "expression construct."
+ "Statement is not a compiled expression construct."
)
elif not self.context.isinsert:
raise exc.InvalidRequestError(
- "Statement is not an insert() " "expression construct."
+ "Statement is not an insert() expression construct."
)
elif self.context.executemany:
return self.context.compiled_parameters
if not self.context.compiled:
raise exc.InvalidRequestError(
- "Statement is not a compiled " "expression construct."
+ "Statement is not a compiled expression construct."
)
elif not self.context.isinsert and not self.context.isupdate:
raise exc.InvalidRequestError(
if not self.context.compiled:
raise exc.InvalidRequestError(
- "Statement is not a compiled " "expression construct."
+ "Statement is not a compiled expression construct."
)
elif not self.context.isinsert and not self.context.isupdate:
raise exc.InvalidRequestError(
and (not self._target_is_object or self._value_is_scalar)
):
raise exc.InvalidRequestError(
- "'any()' not implemented for scalar " "attributes. Use has()."
+ "'any()' not implemented for scalar attributes. Use has()."
)
return self._criterion_exists(
criterion=criterion, is_has=False, **kwargs
or (self._target_is_object and not self._value_is_scalar)
):
raise exc.InvalidRequestError(
- "'has()' not implemented for collections. " "Use any()."
+ "'has()' not implemented for collections. Use any()."
)
return self._criterion_exists(
criterion=criterion, is_has=True, **kwargs
def decorate(fn_or_cls: _F) -> _F:
for type_ in types:
if type_ in _registrars:
- raise AssertionError(
- "Type %s is already " "registered" % type_
- )
+ raise AssertionError("Type %s is already registered" % type_)
_registrars[type_] = fn_or_cls
return fn_or_cls
def _self_inspects(cls: _TT) -> _TT:
if cls in _registrars:
- raise AssertionError("Type %s is already " "registered" % cls)
+ raise AssertionError("Type %s is already registered" % cls)
_registrars[cls] = True
return cls
class ORMStatementRole(roles.StatementRole):
__slots__ = ()
_role_name = (
- "Executable SQL or text() construct, including ORM " "aware objects"
+ "Executable SQL or text() construct, including ORM aware objects"
)
"""
if self.property.uselist:
raise sa_exc.InvalidRequestError(
- "'has()' not implemented for collections. " "Use any()."
+ "'has()' not implemented for collections. Use any()."
)
return self._criterion_exists(criterion, **kwargs)
super().__init__(parent, strategy_key)
if hasattr(self.parent_property, "composite_class"):
raise NotImplementedError(
- "Deferred loading for composite " "types not implemented yet"
+ "Deferred loading for composite types not implemented yet"
)
self.raiseload = self.strategy_opts.get("raiseload", False)
self.columns = self.parent_property.columns
self._equated_columns[c] = self._equated_columns[col]
self.logger.info(
- "%s will use Session.get() to " "optimize instance loads", self
+ "%s will use Session.get() to optimize instance loads", self
)
def init_class_attribute(self, mapper):
self.delete_orphan = "delete-orphan" in values
if self.delete_orphan and not self.delete:
- util.warn(
- "The 'delete-orphan' cascade " "option requires 'delete'."
- )
+ util.warn("The 'delete-orphan' cascade option requires 'delete'.")
return self
def __repr__(self):
def _fallback_column_name(self, column):
raise exc.CompileError(
- "Cannot compile Column object until " "its 'name' is assigned."
+ "Cannot compile Column object until its 'name' is assigned."
)
def visit_lambda_element(self, element, **kw):
def _verify_index_table(self, index):
if index.table is None:
raise exc.CompileError(
- "Index '%s' is not associated " "with any table." % index.name
+ "Index '%s' is not associated with any table." % index.name
)
def visit_create_index(
expr: ColumnElement[Any], op: OperatorType, *arg: Any, **kw: Any
) -> NoReturn:
raise NotImplementedError(
- "Operator '%s' is not supported on " "this expression" % op.__name__
+ "Operator '%s' is not supported on this expression" % op.__name__
)
name = quoted_name(name, quote)
elif quote is not None:
raise exc.ArgumentError(
- "Explicit 'name' is required when " "sending 'quote' argument"
+ "Explicit 'name' is required when sending 'quote' argument"
)
# name = None is expected to be an interim state
action="append",
type=str,
dest="dburi",
- help="Database uri. Multiple OK, " "first one is run by default.",
+ help="Database uri. Multiple OK, first one is run by default.",
)
make_option(
"--dbdriver",
def test_text_rowcount(self, connection):
# test issue #3622, make sure eager rowcount is called for text
result = connection.execute(
- text("update employees set department='Z' " "where department='C'")
+ text("update employees set department='Z' where department='C'")
)
eq_(result.rowcount, 3)
def attrsetter(attrname):
- code = "def set(obj, value):" " obj.%s = value" % attrname
+ code = "def set(obj, value): obj.%s = value" % attrname
env = locals().copy()
exec(code, env)
return env["set"]
t = table("sometable", column("somecolumn"))
self.assert_compile(
t.insert(),
- "INSERT INTO sometable (somecolumn) VALUES " "(:somecolumn)",
+ "INSERT INTO sometable (somecolumn) VALUES (:somecolumn)",
)
def test_update(self):
)
self.assert_compile(
tbl.delete().where(tbl.c.id == 1),
- "DELETE FROM paj.test WHERE paj.test.id = " ":id_1",
+ "DELETE FROM paj.test WHERE paj.test.id = :id_1",
)
s = select(tbl.c.id).where(tbl.c.id == 1)
self.assert_compile(
)
self.assert_compile(
tbl.delete().where(tbl.c.id == 1),
- "DELETE FROM banana.paj.test WHERE " "banana.paj.test.id = :id_1",
+ "DELETE FROM banana.paj.test WHERE banana.paj.test.id = :id_1",
)
s = select(tbl.c.id).where(tbl.c.id == 1)
self.assert_compile(
)
self.assert_compile(
select(func.max(t.c.col1)),
- "SELECT max(sometable.col1) AS max_1 FROM " "sometable",
+ "SELECT max(sometable.col1) AS max_1 FROM sometable",
)
def test_function_overrides(self):
)
d = delete(table1).returning(table1.c.myid, table1.c.name)
self.assert_compile(
- d, "DELETE FROM mytable OUTPUT deleted.myid, " "deleted.name"
+ d, "DELETE FROM mytable OUTPUT deleted.myid, deleted.name"
)
d = (
delete(table1)
)
self.assert_compile(
schema.CreateTable(tbl),
- "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,1)" ")",
+ "CREATE TABLE test (id INTEGER NOT NULL IDENTITY(3,1))",
)
def test_identity_separate_from_primary_key(self):
def test_scalar_strings_named_control(self, scalar_strings, connection):
result = (
connection.exec_driver_sql(
- "SELECT anon_1.my_string " "FROM scalar_strings() AS anon_1"
+ "SELECT anon_1.my_string FROM scalar_strings() AS anon_1"
)
.scalars()
.all()
self.assert_compile(
schema.CreateIndex(idx),
- "CREATE FULLTEXT INDEX test_idx1 " "ON testtbl (data(10))",
+ "CREATE FULLTEXT INDEX test_idx1 ON testtbl (data(10))",
)
def test_create_index_with_text(self):
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):
mysql.MSTimeStamp(),
DefaultClause(
sql.text(
- "'1999-09-09 09:09:09' " "ON UPDATE CURRENT_TIMESTAMP"
+ "'1999-09-09 09:09:09' ON UPDATE CURRENT_TIMESTAMP"
)
),
],
mysql.MSTimeStamp,
DefaultClause(
sql.text(
- "'1999-09-09 09:09:09' " "ON UPDATE CURRENT_TIMESTAMP"
+ "'1999-09-09 09:09:09' ON UPDATE CURRENT_TIMESTAMP"
)
),
],
[
mysql.MSTimeStamp(),
DefaultClause(
- sql.text(
- "CURRENT_TIMESTAMP " "ON UPDATE CURRENT_TIMESTAMP"
- )
+ sql.text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")
),
],
{},
[
mysql.MSTimeStamp,
DefaultClause(
- sql.text(
- "CURRENT_TIMESTAMP " "ON UPDATE CURRENT_TIMESTAMP"
- )
+ sql.text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP")
),
],
{"nullable": False},
t1 = Table("sometable", MetaData(), Column("somecolumn", e1))
self.assert_compile(
schema.CreateTable(t1),
- "CREATE TABLE sometable (somecolumn " "ENUM('x','y','z'))",
+ "CREATE TABLE sometable (somecolumn ENUM('x','y','z'))",
)
t1 = Table(
"sometable",
)
self.assert_compile(
parent.join(child),
- "ed.parent JOIN ed.child ON ed.parent.id = " "ed.child.parent_id",
+ "ed.parent JOIN ed.child ON ed.parent.id = ed.child.parent_id",
)
def test_subquery(self):
q = select(table1.c.name).where(table1.c.name == "foo")
self.assert_compile(
q,
- "SELECT mytable.name FROM mytable WHERE " "mytable.name = :name_1",
+ "SELECT mytable.name FROM mytable WHERE mytable.name = :name_1",
dialect=oracle.dialect(use_ansi=False),
)
)
self.assert_compile(
schema.CreateTable(tbl2),
- "CREATE TABLE testtbl2 (data INTEGER) " "COMPRESS FOR OLTP",
+ "CREATE TABLE testtbl2 (data INTEGER) COMPRESS FOR OLTP",
)
def test_create_index_bitmap_compress(self):
def test_out_params(self, connection):
result = connection.execute(
- text(
- "begin foo(:x_in, :x_out, :y_out, " ":z_out); end;"
- ).bindparams(
+ text("begin foo(:x_in, :x_out, :y_out, :z_out); end;").bindparams(
bindparam("x_in", Float),
outparam("x_out", Integer),
outparam("y_out", Float),
with testing.db.connect() as conn:
eq_(
conn.exec_driver_sql(
- "/*+ this is a comment */ SELECT 1 FROM " "DUAL"
+ "/*+ this is a comment */ SELECT 1 FROM DUAL"
).fetchall(),
[(1,)],
)
)
self.assert_compile(
postgresql.CreateEnumType(e2),
- "CREATE TYPE someschema.somename AS ENUM " "('x', 'y', 'z')",
+ "CREATE TYPE someschema.somename AS ENUM ('x', 'y', 'z')",
)
self.assert_compile(postgresql.DropEnumType(e1), "DROP TYPE somename")
self.assert_compile(
t1 = Table("sometable", MetaData(), Column("somecolumn", e1))
self.assert_compile(
schema.CreateTable(t1),
- "CREATE TABLE sometable (somecolumn " "somename)",
+ "CREATE TABLE sometable (somecolumn somename)",
)
t1 = Table(
"sometable",
self.assert_compile(
schema.CreateIndex(idx),
- "CREATE INDEX test_idx1 ON testtbl " "(data text_pattern_ops)",
+ "CREATE INDEX test_idx1 ON testtbl (data text_pattern_ops)",
dialect=postgresql.dialect(),
)
self.assert_compile(
unique=True,
)
),
- "CREATE UNIQUE INDEX test_idx3 ON test_tbl " "(data3)",
+ "CREATE UNIQUE INDEX test_idx3 ON test_tbl (data3)",
),
(
lambda tbl: schema.CreateIndex(
self.assert_compile(
schema.CreateIndex(idx1),
- "CREATE INDEX test_idx1 ON testtbl " "(data)",
+ "CREATE INDEX test_idx1 ON testtbl (data)",
dialect=postgresql.dialect(),
)
self.assert_compile(
schema.CreateIndex(idx2),
- "CREATE INDEX test_idx2 ON testtbl " "USING btree (data)",
+ "CREATE INDEX test_idx2 ON testtbl USING btree (data)",
dialect=postgresql.dialect(),
)
self.assert_compile(
schema.CreateIndex(idx3),
- "CREATE INDEX test_idx3 ON testtbl " "USING hash (data)",
+ "CREATE INDEX test_idx3 ON testtbl USING hash (data)",
dialect=postgresql.dialect(),
)
self.assert_compile(
schema.CreateIndex(idx1),
- "CREATE INDEX test_idx1 ON testtbl " "(data)",
+ "CREATE INDEX test_idx1 ON testtbl (data)",
)
self.assert_compile(
schema.CreateIndex(idx2),
schema.CreateIndex(
Index("test_idx1", tbl.c.data, postgresql_using="GIST")
),
- "CREATE INDEX test_idx1 ON testtbl " "USING gist (data)",
+ "CREATE INDEX test_idx1 ON testtbl USING gist (data)",
)
self.assert_compile(
self.assert_compile(
schema.CreateIndex(idx1),
- "CREATE INDEX test_idx1 ON testtbl " "(data)",
+ "CREATE INDEX test_idx1 ON testtbl (data)",
dialect=postgresql.dialect(),
)
self.assert_compile(
# default dialect does not, as DBAPIs may be doing this for us
self.assert_compile(
t.update().values({t.c.data[2:5]: [2, 3, 4]}),
- "UPDATE t SET data[%s:%s]=" "%s",
+ "UPDATE t SET data[%s:%s]=%s",
checkparams={"param_1": [2, 3, 4], "data_2": 5, "data_1": 2},
dialect=PGDialect(paramstyle="format"),
)
tbl3 = Table("testtbl3", m, Column("id", Integer), schema="testschema")
stmt = tbl3.select().with_hint(tbl3, "ONLY", "postgresql")
expected = (
- "SELECT testschema.testtbl3.id FROM " "ONLY testschema.testtbl3"
+ "SELECT testschema.testtbl3.id FROM ONLY testschema.testtbl3"
)
self.assert_compile(stmt, expected)
sess = Session()
self.assert_compile(
sess.query(self.table).distinct(),
- "SELECT DISTINCT t.id AS t_id, t.a AS t_a, " "t.b AS t_b FROM t",
+ "SELECT DISTINCT t.id AS t_id, t.a AS t_a, t.b AS t_b FROM t",
)
def test_query_on_columns(self):
"postgresql+psycopg2://USER:PASS@/DB"
"?host=hostA,hostC&port=111,222,333",
),
- ("postgresql+psycopg2://USER:PASS@/DB" "?host=hostA&port=111,222",),
+ ("postgresql+psycopg2://USER:PASS@/DB?host=hostA&port=111,222",),
(
"postgresql+asyncpg://USER:PASS@/DB"
"?host=hostA,hostB,hostC&port=111,333",
if self._strs_render_bind_casts(connection):
self.assert_compile(
matchtable.c.title.match("somstr"),
- "matchtable.title @@ " "plainto_tsquery(%(title_1)s::VARCHAR)",
+ "matchtable.title @@ plainto_tsquery(%(title_1)s::VARCHAR)",
)
else:
self.assert_compile(
)
metadata.create_all(connection)
for stmt in [
- "INSERT INTO bool_table (id, boo) " "VALUES (1, 'false');",
- "INSERT INTO bool_table (id, boo) " "VALUES (2, 'true');",
- "INSERT INTO bool_table (id, boo) " "VALUES (3, '1');",
- "INSERT INTO bool_table (id, boo) " "VALUES (4, '0');",
- "INSERT INTO bool_table (id, boo) " "VALUES (5, 1);",
- "INSERT INTO bool_table (id, boo) " "VALUES (6, 0);",
+ "INSERT INTO bool_table (id, boo) VALUES (1, 'false');",
+ "INSERT INTO bool_table (id, boo) VALUES (2, 'true');",
+ "INSERT INTO bool_table (id, boo) VALUES (3, '1');",
+ "INSERT INTO bool_table (id, boo) VALUES (4, '0');",
+ "INSERT INTO bool_table (id, boo) VALUES (5, 1);",
+ "INSERT INTO bool_table (id, boo) VALUES (6, 0);",
]:
connection.exec_driver_sql(stmt)
@testing.provide_metadata
def test_quoted_identifiers_functional_two(self):
- """ "test the edgiest of edge cases, quoted table/col names
+ """test the edgiest of edge cases, quoted table/col names
that start and end with quotes.
SQLite claims to have fixed this in
),
),
(
- "sqlite:///file:path/to/database?" "mode=ro&uri=true",
+ "sqlite:///file:path/to/database?mode=ro&uri=true",
(
["file:path/to/database?mode=ro"],
{"uri": True, "check_same_thread": False},
self.assert_compile(
schema.CreateColumn(c),
- "test INTEGER NOT NULL " "ON CONFLICT FAIL",
+ "test INTEGER NOT NULL ON CONFLICT FAIL",
dialect=sqlite.dialect(),
)
self.assert_compile(
CreateTable(t),
- "CREATE TABLE n (x VARCHAR(30), " "UNIQUE (x) ON CONFLICT FAIL)",
+ "CREATE TABLE n (x VARCHAR(30), UNIQUE (x) ON CONFLICT FAIL)",
dialect=sqlite.dialect(),
)
(
"foo1=bar1&foo2=bar21&foo2=bar22&foo3=bar31",
"foo2=bar23&foo3=bar32&foo3=bar33",
- "foo1=bar1&foo2=bar23&" "foo3=bar32&foo3=bar33",
+ "foo1=bar1&foo2=bar23&foo3=bar32&foo3=bar33",
False,
),
)
e = engine_from_config(config, module=dbapi, _initialize=False)
assert e.pool._recycle == 50
assert e.url == url.make_url(
- "postgresql+psycopg2://scott:tiger@somehost/test?foo" "z=somevalue"
+ "postgresql+psycopg2://scott:tiger@somehost/test?fooz=somevalue"
)
assert e.echo is True
assert not trans.is_active
eq_(
- connection.exec_driver_sql(
- "select count(*) from " "users"
- ).scalar(),
+ connection.exec_driver_sql("select count(*) from users").scalar(),
2,
)
connection.rollback()
assert_raises_message(
sa.exc.ArgumentError,
- "Can't add additional column 'foo' when " "specifying __table__",
+ "Can't add additional column 'foo' when specifying __table__",
go,
)
assert_raises_message(
exc.InvalidRequestError,
- "'addresses' is not an instance of " "ColumnProperty",
+ "'addresses' is not an instance of ColumnProperty",
configure_mappers,
)
assert_raises_message(
AttributeError,
- "does not have a mapped column named " "'__table__'",
+ "does not have a mapped column named '__table__'",
configure_mappers,
)
def test_oops(self):
with testing.expect_warnings(
- "Ignoring declarative-like tuple value of " "attribute 'name'"
+ "Ignoring declarative-like tuple value of attribute 'name'"
):
class User(Base, ComparableEntity):
assert_raises_message(
sa.exc.ArgumentError,
- "Can't add additional column 'tada' when " "specifying __table__",
+ "Can't add additional column 'tada' when specifying __table__",
go,
)
asserter.assert_(
CompiledSQL(
- "UPDATE users SET name=:name WHERE " "users.id = :users_id",
+ "UPDATE users SET name=:name WHERE users.id = :users_id",
[
{"users_id": 1, "name": "u1new"},
{"users_id": 2, "name": "u2"},
s.flush()
asserter.assert_(
RegexSQL(
- "SELECT .* " "FROM c WHERE :param_1 = c.bid", [{"param_1": 3}]
+ "SELECT .* FROM c WHERE :param_1 = c.bid", [{"param_1": 3}]
),
CompiledSQL("DELETE FROM c WHERE c.cid = :cid", [{"cid": 1}]),
CompiledSQL("DELETE FROM b WHERE b.id = :id", [{"id": 3}]),
)
def test_optimized_passes(self):
- """ "test that the 'optimized load' routine doesn't crash when
+ """test that the 'optimized load' routine doesn't crash when
a column in the join condition is not available."""
base, sub = self.tables.base, self.tables.sub
__mapper_args__ = {"polymorphic_identity": "b"}
with expect_warnings(
- r"Mapper\[C\(a\)\] does not indicate a " "'polymorphic_identity',"
+ r"Mapper\[C\(a\)\] does not indicate a 'polymorphic_identity',"
):
class C(A):
engine = {"e1": e1, "e2": e2, "e3": e3}[expected_engine_name]
with mock.patch(
- "sqlalchemy.orm.context." "ORMCompileState.orm_setup_cursor_result"
+ "sqlalchemy.orm.context.ORMCompileState.orm_setup_cursor_result"
), mock.patch(
"sqlalchemy.orm.context.ORMCompileState.orm_execute_statement"
), mock.patch(
assert_raises_message(
sa.exc.InvalidRequestError,
- "Session already has a Connection " "associated",
+ "Session already has a Connection associated",
transaction._connection_for_bind,
testing.db.connect(),
None,
self.assert_compile(
stmt,
- "DELETE FROM users AS users_1 " "WHERE users_1.name = :name_1",
+ "DELETE FROM users AS users_1 WHERE users_1.name = :name_1",
)
@testing.variation("stmt_type", ["core", "orm"])
],
),
CompiledSQL(
- "DELETE FROM person " "WHERE person.id = :id",
+ "DELETE FROM person WHERE person.id = :id",
lambda ctx: [{"id": p.id}],
),
CompiledSQL(
@testing.fails_on("mssql", "FIXME: unknown")
@testing.fails_on(
- "oracle", "Oracle doesn't support boolean expressions as " "columns"
+ "oracle", "Oracle doesn't support boolean expressions as columns"
)
@testing.fails_on(
"postgresql+pg8000",
class DeferredMapperEventsTest(RemoveORMEventsGlobally, _fixtures.FixtureTest):
- """ "test event listeners against unmapped classes.
+ """test event listeners against unmapped classes.
This incurs special logic. Note if we ever do the "remove" case,
it has to get all of these, too.
self.assert_compile(
stmt,
- "SELECT users.id, " "users.name " "FROM users",
+ "SELECT users.id, users.name FROM users",
)
is_true(um.configured)
if first_element
else (Load(Item).joinedload(Keyword),)
),
- "expected ORM mapped attribute for loader " "strategy argument",
+ "expected ORM mapped attribute for loader strategy argument",
)
@testing.combinations(
self.assert_compile(
q1,
- "SELECT users.id AS foo FROM users " "WHERE users.name = :name_1",
+ "SELECT users.id AS foo FROM users WHERE users.name = :name_1",
)
def test_empty_filters(self):
q1 = sess.query(User)
self.assert_compile(
sess.query(q1.exists()),
- "SELECT EXISTS (" "SELECT 1 FROM users" ") AS anon_1",
+ "SELECT EXISTS (SELECT 1 FROM users) AS anon_1",
)
q2 = sess.query(User).filter(User.name == "fred")
q1 = sess.query(User.id)
self.assert_compile(
sess.query(q1.exists()),
- "SELECT EXISTS (" "SELECT 1 FROM users" ") AS anon_1",
+ "SELECT EXISTS (SELECT 1 FROM users) AS anon_1",
)
def test_exists_labeled_col_expression(self):
q1 = sess.query(User.id.label("foo"))
self.assert_compile(
sess.query(q1.exists()),
- "SELECT EXISTS (" "SELECT 1 FROM users" ") AS anon_1",
+ "SELECT EXISTS (SELECT 1 FROM users) AS anon_1",
)
def test_exists_arbitrary_col_expression(self):
q1 = sess.query(func.foo(User.id))
self.assert_compile(
sess.query(q1.exists()),
- "SELECT EXISTS (" "SELECT 1 FROM users" ") AS anon_1",
+ "SELECT EXISTS (SELECT 1 FROM users) AS anon_1",
)
def test_exists_col_warning(self):
User = self.classes.User
sess = fixture_session()
query = sess.query(User.name).prefix_with("PREFIX_1")
- expected = "SELECT PREFIX_1 " "users.name AS users_name FROM users"
+ expected = "SELECT PREFIX_1 users.name AS users_name FROM users"
self.assert_compile(query, expected, dialect=default.DefaultDialect())
def test_one_suffix(self):
sess = fixture_session()
query = sess.query(User.name).prefix_with("PREFIX_1", "PREFIX_2")
expected = (
- "SELECT PREFIX_1 PREFIX_2 " "users.name AS users_name FROM users"
+ "SELECT PREFIX_1 PREFIX_2 users.name AS users_name FROM users"
)
self.assert_compile(query, expected, dialect=default.DefaultDialect())
testing.db,
q.all,
CompiledSQL(
- "SELECT a.id AS a_id, a.q AS a_q " "FROM a ORDER BY a.id", [{}]
+ "SELECT a.id AS a_id, a.q AS a_q FROM a ORDER BY a.id", [{}]
),
# in the very unlikely case that the the FK col on parent is
# deferred, we degrade to the JOIN version so that we don't need to
testing.db,
session.flush,
CompiledSQL(
- "INSERT INTO users (name) " "VALUES (:name)",
+ "INSERT INTO users (name) VALUES (:name)",
{"name": "imnewlyadded"},
),
AllOf(
{"description": "item4updated", "items_id": objects[4].id},
),
CompiledSQL(
- "INSERT INTO keywords (name) " "VALUES (:name)",
+ "INSERT INTO keywords (name) VALUES (:name)",
{"name": "yellow"},
),
CompiledSQL(
# sync operation during _save_obj().update, this is safe to remove
# again.
CompiledSQL(
- "UPDATE child SET pid=:pid " "WHERE child.cid = :child_cid",
+ "UPDATE child SET pid=:pid WHERE child.cid = :child_cid",
{"pid": 1, "child_cid": 1},
),
)
testing.db,
s.flush,
CompiledSQL(
- "INSERT INTO test2 (id, foo, bar) " "VALUES (:id, :foo, :bar)",
+ "INSERT INTO test2 (id, foo, bar) VALUES (:id, :foo, :bar)",
[{"id": 1, "foo": None, "bar": 2}],
),
)
print("Total cpu seconds: %.2f" % stats.total_tt)
print(
"Total execute calls: %d"
- % counts_by_methname[
- "<method 'execute' of 'sqlite3.Cursor' " "objects>"
- ]
+ % counts_by_methname["<method 'execute' of 'sqlite3.Cursor' objects>"]
)
print(
"Total executemany calls: %d"
% counts_by_methname.get(
- "<method 'executemany' of 'sqlite3.Cursor' " "objects>", 0
+ "<method 'executemany' of 'sqlite3.Cursor' objects>", 0
)
)
@property
def emulated_lastrowid(self):
- """ "target dialect retrieves cursor.lastrowid or an equivalent
+ """target dialect retrieves cursor.lastrowid or an equivalent
after an insert() construct executes.
"""
return fails_on_everything_except(
@property
def emulated_lastrowid_even_with_sequences(self):
- """ "target dialect retrieves cursor.lastrowid or an equivalent
+ """target dialect retrieves cursor.lastrowid or an equivalent
after an insert() construct executes, even if the table has a
Sequence on it.
"""
@property
def dbapi_lastrowid(self):
- """ "target backend includes a 'lastrowid' accessor on the DBAPI
+ """target backend includes a 'lastrowid' accessor on the DBAPI
cursor object.
"""
)
self.assert_compile(
select(select(table1.c.name).label("foo")),
- "SELECT (SELECT mytable.name FROM mytable) " "AS foo",
+ "SELECT (SELECT mytable.name FROM mytable) AS foo",
)
# scalar selects should not have any attributes on their 'c' or
self.assert_compile(
s3,
- "SELECT NULL AS anon_1, NULL AS anon__1 " "UNION "
+ "SELECT NULL AS anon_1, NULL AS anon__1 UNION "
# without the feature tested in test_deduping_hash_algo we'd get
# "SELECT true AS anon_2, true AS anon__1",
"SELECT true AS anon_2, true AS anon__2",
)
assert_raises_message(
exc.CompileError,
- "conflicts with unique bind parameter " "of the same name",
+ "conflicts with unique bind parameter of the same name",
str,
s,
)
)
assert_raises_message(
exc.CompileError,
- "conflicts with unique bind parameter " "of the same name",
+ "conflicts with unique bind parameter of the same name",
str,
s,
)
)
self.assert_compile(
expr,
- "(mytable.myid, mytable.name) IN " "(__[POSTCOMPILE_param_1])",
+ "(mytable.myid, mytable.name) IN (__[POSTCOMPILE_param_1])",
checkparams={"param_1": [(1, "foo"), (5, "bar")]},
check_post_param={"param_1": [(1, "foo"), (5, "bar")]},
check_literal_execute={},
dialect.tuple_in_values = True
self.assert_compile(
tuple_(table1.c.myid, table1.c.name).in_([(1, "foo"), (5, "bar")]),
- "(mytable.myid, mytable.name) IN " "(__[POSTCOMPILE_param_1])",
+ "(mytable.myid, mytable.name) IN (__[POSTCOMPILE_param_1])",
dialect=dialect,
checkparams={"param_1": [(1, "foo"), (5, "bar")]},
check_post_param={"param_1": [(1, "foo"), (5, "bar")]},
select(table1.c.myid).where(
table1.c.myid == bindparam("foo", 5, literal_execute=True)
),
- "SELECT mytable.myid FROM mytable " "WHERE mytable.myid = 5",
+ "SELECT mytable.myid FROM mytable WHERE mytable.myid = 5",
literal_binds=True,
)
select(table1.c.myid).where(
table1.c.myid == bindparam("foo", 5, literal_execute=True)
),
- "SELECT mytable.myid FROM mytable " "WHERE mytable.myid = 5",
+ "SELECT mytable.myid FROM mytable WHERE mytable.myid = 5",
render_postcompile=True,
)
eq_ignore_whitespace(
str(schema.AddConstraint(cons)),
- "ALTER TABLE testtbl ADD EXCLUDE USING gist " "(room WITH =)",
+ "ALTER TABLE testtbl ADD EXCLUDE USING gist (room WITH =)",
)
def test_try_cast(self):
s = select(t1.c.a)
s2 = select(t1).where(t1.c.a == s.scalar_subquery())
self.assert_compile(
- s2, "SELECT t1.a FROM t1 WHERE t1.a = " "(SELECT t1.a FROM t1)"
+ s2, "SELECT t1.a FROM t1 WHERE t1.a = (SELECT t1.a FROM t1)"
)
def test_correlate_semiauto_where_singlefrom(self):
if auto:
fk_assertions.append(
CompiledSQL(
- "ALTER TABLE a ADD " "FOREIGN KEY(bid) REFERENCES b (id)"
+ "ALTER TABLE a ADD FOREIGN KEY(bid) REFERENCES b (id)"
)
)
assertions.append(AllOf(*fk_assertions))
),
AllOf(
CompiledSQL(
- "ALTER TABLE b ADD " "FOREIGN KEY(aid) REFERENCES a (id)"
+ "ALTER TABLE b ADD FOREIGN KEY(aid) REFERENCES a (id)"
),
CompiledSQL(
- "ALTER TABLE a ADD " "FOREIGN KEY(bid) REFERENCES b (id)"
+ "ALTER TABLE a ADD FOREIGN KEY(bid) REFERENCES b (id)"
),
),
]
RegexSQL("^CREATE TABLE events"),
AllOf(
CompiledSQL(
- "CREATE UNIQUE INDEX ix_events_name ON events " "(name)"
+ "CREATE UNIQUE INDEX ix_events_name ON events (name)"
),
CompiledSQL(
- "CREATE INDEX ix_events_location ON events " "(location)"
+ "CREATE INDEX ix_events_location ON events (location)"
),
CompiledSQL(
"CREATE UNIQUE INDEX sport_announcer ON events "
self.assert_compile(
schema.CreateIndex(ix1),
- "CREATE INDEX %s " "ON %s (%s)" % (exp, tname, cname),
+ "CREATE INDEX %s ON %s (%s)" % (exp, tname, cname),
dialect=dialect,
)
# is disabled
self.assert_compile(
schema.CreateTable(t),
- "CREATE TABLE tbl (" "a INTEGER, " "b INTEGER" ")",
+ "CREATE TABLE tbl (a INTEGER, b INTEGER)",
)
def test_render_drop_constraint(self):
else:
assert_raises_message(
CompileError,
- "Multiple, unrelated CTEs found " "with the same name: 'cte1'",
+ "Multiple, unrelated CTEs found with the same name: 'cte1'",
s.compile,
)
sel = select(basefrom.c.a)
with testing.expect_deprecated(
- r"The Selectable.replace_selectable\(\) " "method is deprecated"
+ r"The Selectable.replace_selectable\(\) method is deprecated"
):
replaced = sel.replace_selectable(
basefrom, basefrom.join(joinfrom, basefrom.c.a == joinfrom.c.a)
def test_table_to_alias_9(self):
s = select(literal_column("*")).select_from(t1).alias("foo")
self.assert_compile(
- s.select(), "SELECT foo.* FROM (SELECT * FROM table1) " "AS foo"
+ s.select(), "SELECT foo.* FROM (SELECT * FROM table1) AS foo"
)
def test_table_to_alias_10(self):
vis = sql_util.ClauseAdapter(t1alias)
self.assert_compile(
vis.traverse(s.select()),
- "SELECT foo.* FROM (SELECT * FROM table1 " "AS t1alias) AS foo",
+ "SELECT foo.* FROM (SELECT * FROM table1 AS t1alias) AS foo",
)
def test_table_to_alias_11(self):
s = select(literal_column("*")).select_from(t1).alias("foo")
self.assert_compile(
- s.select(), "SELECT foo.* FROM (SELECT * FROM table1) " "AS foo"
+ s.select(), "SELECT foo.* FROM (SELECT * FROM table1) AS foo"
)
def test_table_to_alias_12(self):
ff = vis.traverse(func.count(t1.c.col1).label("foo"))
self.assert_compile(
select(ff),
- "SELECT count(t1alias.col1) AS foo FROM " "table1 AS t1alias",
+ "SELECT count(t1alias.col1) AS foo FROM table1 AS t1alias",
)
assert list(_from_objects(ff)) == [t1alias]
)
self.assert_compile(
sql_util.splice_joins(table1, j2),
- "table1 JOIN table4 AS table4_1 ON " "table1.col3 = table4_1.col3",
+ "table1 JOIN table4 AS table4_1 ON table1.col3 = table4_1.col3",
)
self.assert_compile(
sql_util.splice_joins(sql_util.splice_joins(table1, j1), j2),
def test_columns(self):
s = t1.select()
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
select_copy = s.add_columns(column("yyy"))
self.assert_compile(
select_copy,
- "SELECT table1.col1, table1.col2, " "table1.col3, yyy FROM table1",
+ "SELECT table1.col1, table1.col2, table1.col3, yyy FROM table1",
)
is_not(s.selected_columns, select_copy.selected_columns)
is_not(s._raw_columns, select_copy._raw_columns)
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
def test_froms(self):
s = t1.select()
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
select_copy = s.select_from(t2)
self.assert_compile(
)
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
def test_prefixes(self):
s = t1.select()
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
select_copy = s.prefix_with("FOOBER")
self.assert_compile(
"table1.col3 FROM table1",
)
self.assert_compile(
- s, "SELECT table1.col1, table1.col2, " "table1.col3 FROM table1"
+ s, "SELECT table1.col1, table1.col2, table1.col3 FROM table1"
)
def test_execution_options(self):
Column("q", Integer),
)
with expect_warnings(
- "Column 't.x' is marked as a member.*" "may not store NULL.$"
+ "Column 't.x' is marked as a member.*may not store NULL.$"
):
self.assert_compile(
t.insert(), "INSERT INTO t (q) VALUES (:q)", params={"q": 5}
d = postgresql.dialect()
d.implicit_returning = True
with expect_warnings(
- "Column 't.x' is marked as a member.*" "may not store NULL.$"
+ "Column 't.x' is marked as a member.*may not store NULL.$"
):
self.assert_compile(
t.insert(),
d.implicit_returning = False
with expect_warnings(
- "Column 't.x' is marked as a member.*" "may not store NULL.$"
+ "Column 't.x' is marked as a member.*may not store NULL.$"
):
self.assert_compile(
t.insert(),
Column("notpk", String(10), nullable=True),
)
with expect_warnings(
- "Column 't.id' is marked as a member.*" "may not store NULL.$"
+ "Column 't.id' is marked as a member.*may not store NULL.$"
):
self.assert_compile(
t.insert(),
self.assert_compile(
stmt,
- "INSERT INTO sometable (id, data) VALUES " "(foobar(), ?)",
+ "INSERT INTO sometable (id, data) VALUES (foobar(), ?)",
checkparams={"data": "foo"},
params={"data": "foo"},
dialect=dialect,
self.assert_compile(
go("u1"),
- "SELECT users.id FROM users " "WHERE users.name = 'u1'",
+ "SELECT users.id FROM users WHERE users.name = 'u1'",
literal_binds=True,
)
assert_raises_message(
exc.InvalidRequestError,
- "Table 'users' is already defined for this " "MetaData instance.",
+ "Table 'users' is already defined for this MetaData instance.",
go,
)
dialect.max_identifier_length = 15
self.assert_compile(
schema.CreateIndex(ix),
- "CREATE INDEX ix_user_2de9 ON " '"user" (data, "Data2", "Data3")',
+ 'CREATE INDEX ix_user_2de9 ON "user" (data, "Data2", "Data3")',
dialect=dialect,
)
# no issue with native boolean
self.assert_compile(
schema.CreateTable(u1),
- 'CREATE TABLE "user" (' "x BOOLEAN" ")",
+ """CREATE TABLE "user" (x BOOLEAN)""",
dialect="postgresql",
)
),
(
lambda p, q: (1 - p) * (2 - q) * (3 - p) * (4 - q),
- "(:p_1 - t.p) * (:q_1 - t.q) * " "(:p_2 - t.p) * (:q_2 - t.q)",
+ "(:p_1 - t.p) * (:q_1 - t.q) * (:p_2 - t.p) * (:q_2 - t.q)",
),
(
lambda p, q: (
self.table.c.myid.match("foo"),
self.table.c.myid.regexp_match("xx"),
),
- "mytable.myid MATCH :myid_1 AND " "mytable.myid <regexp> :myid_2",
+ "mytable.myid MATCH :myid_1 AND mytable.myid <regexp> :myid_2",
)
self.assert_compile(
and_(
# what if table/schema *are* quoted?
self.assert_compile(
t1.select().set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL),
- "SELECT " "Foo.T1.Col1 AS Foo_T1_Col1 " "FROM " "Foo.T1",
+ "SELECT Foo.T1.Col1 AS Foo_T1_Col1 FROM Foo.T1",
)
def test_quote_flag_propagate_check_constraint(self):
CheckConstraint(t.c.x > 5)
self.assert_compile(
schema.CreateTable(t),
- "CREATE TABLE t (" '"x" INTEGER, ' 'CHECK ("x" > 5)' ")",
+ 'CREATE TABLE t ("x" INTEGER, CHECK ("x" > 5))',
)
def test_quote_flag_propagate_index(self):
if use_pickle:
with expect_raises_message(
exc.NoSuchColumnError,
- "Row was unpickled; lookup by ColumnElement is " "unsupported",
+ "Row was unpickled; lookup by ColumnElement is unsupported",
):
result[0]._mapping[users.c.user_id]
else:
if use_pickle:
with expect_raises_message(
exc.NoSuchColumnError,
- "Row was unpickled; lookup by ColumnElement is " "unsupported",
+ "Row was unpickled; lookup by ColumnElement is unsupported",
):
result[0]._mapping[users.c.user_name]
else:
r"SELECT * FROM pg_attribute WHERE "
r"attrelid = :tab\:\:regclass"
),
- "SELECT * FROM pg_attribute WHERE " "attrelid = %(tab)s::regclass",
+ "SELECT * FROM pg_attribute WHERE attrelid = %(tab)s::regclass",
params={"tab": None},
dialect="postgresql",
)
r"SELECT * FROM pg_attribute WHERE "
r"attrelid = foo::regclass"
),
- "SELECT * FROM pg_attribute WHERE " "attrelid = foo::regclass",
+ "SELECT * FROM pg_attribute WHERE attrelid = foo::regclass",
params={},
dialect="postgresql",
)
assert_raises(
(exc.DBAPIError,),
connection.exec_driver_sql,
- "insert into my_table " "(data) values('four')",
+ "insert into my_table (data) values('four')",
)
trans.rollback()
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
