我们从Python开源项目中,提取了以下50个代码示例,用于说明如何使用sqlalchemy.literal()。
def define_tables(cls, metadata): Table('autoinc_pk', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('data', String(50)) ) Table('manual_pk', metadata, Column('id', Integer, primary_key=True, autoincrement=False), Column('data', String(50)) ) Table('includes_defaults', metadata, Column('id', Integer, primary_key=True, test_needs_autoincrement=True), Column('data', String(50)), Column('x', Integer, default=5), Column('y', Integer, default=literal_column("2", type_=Integer) + literal(2)))
def test_escape_binary_mod(self): query = session.query(literal(1) % literal(2)) self.assertEqual( self.compile(query, literal_binds=True), 'SELECT 1 %% 2 AS anon_1' ) table = Table( 't', self.metadata(), Column('x', types.Int32, primary_key=True), engines.Memory() ) query = session.query(table.c.x % table.c.x) self.assertEqual( self.compile(query, literal_binds=True), 'SELECT x %% x AS anon_1 FROM t' )
def exists(self): """A convenience method that turns a query into an EXISTS subquery of the form EXISTS (SELECT 1 FROM ... WHERE ...). e.g.:: q = session.query(User).filter(User.name == 'fred') session.query(q.exists()) Producing SQL similar to:: SELECT EXISTS ( SELECT 1 FROM users WHERE users.name = :name_1 ) AS anon_1 The EXISTS construct is usually used in the WHERE clause:: session.query(User.id).filter(q.exists()).scalar() Note that some databases such as SQL Server don't allow an EXISTS expression to be present in the columns clause of a SELECT. To select a simple boolean value based on the exists as a WHERE, use :func:`.literal`:: from sqlalchemy import literal session.query(literal(True)).filter(q.exists()).scalar() .. versionadded:: 0.8.1 """ # .add_columns() for the case that we are a query().select_from(X), # so that ".statement" can be produced (#2995) but also without # omitting the FROM clause from a query(X) (#2818); # .with_only_columns() after we have a core select() so that # we get just "SELECT 1" without any entities. return sql.exists(self.add_columns('1').with_labels(). statement.with_only_columns([1]))
def exists(self): """A convenience method that turns a query into an EXISTS subquery of the form EXISTS (SELECT 1 FROM ... WHERE ...). e.g.:: q = session.query(User).filter(User.name == 'fred') session.query(q.exists()) Producing SQL similar to:: SELECT EXISTS ( SELECT 1 FROM users WHERE users.name = :name_1 ) AS anon_1 The EXISTS construct is usually used in the WHERE clause:: session.query(User.id).filter(q.exists()).scalar() Note that some databases such as SQL Server don't allow an EXISTS expression to be present in the columns clause of a SELECT. To select a simple boolean value based on the exists as a WHERE, use :func:`.literal`:: from sqlalchemy import literal session.query(literal(True)).filter(q.exists()).scalar() .. versionadded:: 0.8.1 """ # .add_columns() for the case that we are a query().select_from(X), # so that ".statement" can be produced (#2995) but also without # omitting the FROM clause from a query(X) (#2818); # .with_only_columns() after we have a core select() so that # we get just "SELECT 1" without any entities. return sql.exists(self.add_columns('1').with_labels(). statement.with_only_columns(['1']))
def _do_always_true(self, clause_name, payload): """We just ignore the payload.""" from sqlalchemy import literal return literal(True)
def _do_always_false(self, clause_name, payload): """We just ignore the payload.""" from sqlalchemy import literal return literal(False) # Searches for observing sessions
def for_filling(self, dialect): # NOTE(sileht): This must be used only for patching resource type # to fill all row with a default value and then switch back the # server_default to None if self.fill is None: return None # NOTE(sileht): server_default must be converted in sql element return sqlalchemy.literal(self.fill)
def for_filling(self, dialect): if self.fill is None: return False # Don't set any server_default return sqlalchemy.literal( self.satype.process_bind_param(self.fill, dialect))
def test_lambda_functions(self): query = session.query( func.arrayFilter( Lambda(lambda x: x.like('%World%')), literal(['Hello', 'abc World'], types.Array(types.String)) ).label('test') ) self.assertEqual( self.compile(query, literal_binds=True), "SELECT arrayFilter(" "x -> x LIKE '%%World%%', ['Hello', 'abc World']" ") AS test" )
def test_unsupported_expressoin(self): t1, t2 = self.create_tables(2) query = session.query(t1.c.x).join(t2, literal(True), any=True) with self.assertRaises(exc.CompileError) as ex: self.compile(query) self.assertIn('Only tuple elements are supported', str(ex.exception))
def test_if(self): expression = func.if_( literal(1) > literal(2), text('a'), text('b'), ) self.assertEqual( self.compile(expression, literal_binds=True), 'if(1 > 2, a, b)' )
def test_else_required(self): with self.assertRaises(exc.CompileError) as ex: self.compile(case([(literal(1), 0)])) self.assertEqual( str(ex.exception), 'ELSE clause is required in CASE' )
def test_case(self): expression = case([(literal(1), 0)], else_=1) self.assertEqual( self.compile(expression, literal_binds=True), 'CASE WHEN 1 THEN 0 ELSE 1 END' )
def get_params(self): params = super(GraphiteMergeTree, self).get_params() params.append(literal(self.config_name)) return params