users = Table('users', db,
Column('user_id', Integer, primary_key = True),
Column('user_name', String(40)),
- Column('fullname', String(100)),
- Column('email_address', String(80))
+ Column('password', String(80))
)
# a table that stores mailing addresses associated with a specific user
# then, call execute on the Select object:
<&formatting.myt:poplink&>c = s.execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address FROM users
+SELECT users.user_id, users.user_name, users.password FROM users
{}
</&>
# the SQL text of any clause object can also be viewed via the str() call:
>>> str(s)
- SELECT users.user_id, users.user_name, users.fullname, users.email_address FROM users
+ SELECT users.user_id, users.user_name, users.password FROM users
</&>
<p>The object returned by the execute call is a <span class="codeline">sqlalchemy.engine.ResultProxy</span> object, which acts very much like a DBAPI <span class="codeline">cursor</span> object in the context of a result set, except that the rows returned can address their columns by ordinal position, column name, or even column object:</p>
# select rows, get resulting ResultProxy object
<&formatting.myt:poplink&>c = users.select().execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address FROM users
+SELECT users.user_id, users.user_name, users.password FROM users
{}
</&>
# get one row
user_name = row['user_name']
# or column object
- fullname = row[users.c.fullname]
+ password = row[users.c.password]
# rowproxy object also supports fetchall()
rows = c.fetchall()
# full tables
<&formatting.myt:poplink&>c = select([users, addresses]).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname,
-users.email_address, addresses.address_id, addresses.user_id,
+SELECT users.user_id, users.user_name, users.password,
+addresses.address_id, addresses.user_id,
addresses.street, addresses.city, addresses.state, addresses.zip
FROM users, addresses
{}
# combinations
<&formatting.myt:poplink&>c = select([users, addresses.c.zip]).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname,
-users.email_address, addresses.zip FROM users, addresses
+SELECT users.user_id, users.user_name, users.password,
+addresses.zip FROM users, addresses
{}
</&>
</&>
<&|formatting.myt:code&>
<&formatting.myt:poplink&>c = users.select(users.c.user_id == 7).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname,
-users.email_address FROM users
-WHERE users.user_id = :users_user_id
+SELECT users.user_id, users.user_name, users.password,
+FROM users WHERE users.user_id = :users_user_id
{'users_user_id': 7}
</&>
</&>
# another comparison operator
<&formatting.myt:poplink&>c = select([users], users.c.user_id>7).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname,
-users.email_address FROM users WHERE users.user_id > :users_user_id
+SELECT users.user_id, users.user_name, users.password,
+FROM users WHERE users.user_id > :users_user_id
{'users_user_id': 7}
</&>
# OR keyword
<&formatting.myt:poplink&>c = users.select(or_(users.c.user_name=='jack', users.c.user_name=='ed')).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
+SELECT users.user_id, users.user_name, users.password
FROM users WHERE users.user_name = :users_user_name
OR users.user_name = :users_user_name_1
{'users_user_name_1': 'ed', 'users_user_name': 'jack'}
</&>
# AND keyword
-<&formatting.myt:poplink&>c = users.select(and_(users.c.user_name=='jack', users.c.fullname=='ed')).execute()
+<&formatting.myt:poplink&>c = users.select(and_(users.c.user_name=='jack', users.c.password=='dog')).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
+SELECT users.user_id, users.user_name, users.password
FROM users WHERE users.user_name = :users_user_name
-AND users.fullname = :users_fullname
-{'users_user_name': 'jack', 'users_fullname': 'ed'}
+AND users.password = :users_password
+{'users_user_name': 'jack', 'users_password': 'dog'}
</&>
# IN clause
<&formatting.myt:poplink&>c = users.select(users.c.user_name.in_('jack', 'ed', 'fred')).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
+SELECT users.user_id, users.user_name, users.password
FROM users WHERE users.user_name IN ('jack', 'ed', 'fred')
</&>
# join users and addresses together
<&formatting.myt:poplink&>c = select([users, addresses], users.c.user_id==addresses.c.address_id).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address,
+SELECT users.user_id, users.user_name, users.password,
addresses.address_id, addresses.user_id, addresses.street, addresses.city,
addresses.state, addresses.zip
FROM users, addresses
users.c.user_name=='fred'
)).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
+SELECT users.user_id, users.user_name, users.password
FROM users, addresses WHERE users.user_id = addresses.user_id
AND users.user_name = :users_user_name
{'users_user_name': 'fred'}
<P>The ORDER BY clause of a select statement can be specified as individual columns to order by within an array specified via the <span class="codeline">order_by</span> parameter, and optional usage of the asc() and desc() functions:
<&|formatting.myt:code &>
# straight order by
-<&formatting.myt:poplink&>c = users.select(order_by=[users.c.fullname]).execute()
+<&formatting.myt:poplink&>c = users.select(order_by=[users.c.user_name]).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
-FROM users ORDER BY users.fullname
+SELECT users.user_id, users.user_name, users.password
+FROM users ORDER BY users.user_name
</&>
# descending/ascending order by on multiple columns
<&formatting.myt:poplink&>c = users.select(
users.c.user_name>'J',
- order_by=[desc(users.c.fullname), asc(users.c.user_name)]).execute()
+ order_by=[desc(users.c.user_id), asc(users.c.user_name)]).execute()
<&|formatting.myt:codepopper, link="sql" &>
-SELECT users.user_id, users.user_name, users.fullname, users.email_address
+SELECT users.user_id, users.user_name, users.password
FROM users WHERE users.user_name > :users_user_name
-ORDER BY users.fullname DESC, users.user_name ASC
+ORDER BY users.user_id DESC, users.user_name ASC
{'users_user_name':'J'}
</&>
</&>
schema = 'remote_owner'
)
+users = Table('users', db,
+ Column('user_id', Integer, primary_key = True),
+ Column('user_name', String(40)),
+ Column('password', String(10)),
+)
+
+addresses = Table('addresses', db,
+ Column('address_id', Integer, primary_key = True),
+ Column('user_id', Integer, ForeignKey("users.user_id")),
+ Column('street', String(100)),
+ Column('city', String(80)),
+ Column('state', String(2)),
+ Column('zip', String(10))
+)
+
+
class SQLTest(PersistTest):
def runtest(self, clause, result, engine = None, params = None, checkparams = None):
c = clause.compile(engine, params)
myothertable.othername FROM mytable, myothertable")
def testsubquery(self):
- s = select([table], table.c.name == 'jack')
- self.runtest(
- select(
- [s],
- s.c.id == 7
- )
- ,
- "SELECT myid, name, description FROM (SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.name = :mytable_name) WHERE myid = :myid")
+# s = select([table], table.c.name == 'jack')
+ # self.runtest(
+ # select(
+ # [s],
+# s.c.id == 7
+# )
+ # ,
+ # "SELECT myid, name, description FROM (SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.name = :mytable_name) WHERE myid = :myid")
- sq = Select([table])
+ sq = select([table])
self.runtest(
sq.select(),
"SELECT myid, name, description FROM (SELECT mytable.myid, mytable.name, mytable.description FROM mytable)"
def testcorrelatedsubquery(self):
self.runtest(
- select([table], table.c.id == select([table2.c.id], table.c.name == table2.c.name)),
+ table.select(table.c.id == select([table2.c.id], table.c.name == table2.c.name)),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE mytable.myid = (SELECT myothertable.otherid FROM myothertable WHERE mytable.name = myothertable.othername)"
)
self.runtest(
- select([table], exists([1], table2.c.id == table.c.id)),
+ table.select(exists([1], table2.c.id == table.c.id)),
"SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)"
)
- s = subquery('sq2', [table], exists([1], table2.c.id == table.c.id))
+ talias = table.alias('ta')
+ s = subquery('sq2', [talias], exists([1], table2.c.id == talias.c.id))
self.runtest(
select([s, table])
- ,"SELECT sq2.myid, sq2.name, sq2.description, mytable.myid, mytable.name, mytable.description FROM (SELECT mytable.myid, mytable.name, mytable.description FROM mytable WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = mytable.myid)) sq2, mytable")
+ ,"SELECT sq2.myid, sq2.name, sq2.description, mytable.myid, mytable.name, mytable.description FROM (SELECT ta.myid, ta.name, ta.description FROM mytable ta WHERE EXISTS (SELECT 1 FROM myothertable WHERE myothertable.otherid = ta.myid)) sq2, mytable")
+
+ s = select([addresses.c.street], addresses.c.user_id==users.c.user_id).alias('s')
+ self.runtest(
+ select([users, s.c.street], from_obj=[s]),
+ """SELECT users.user_id, users.user_name, users.password, s.street FROM users, (SELECT addresses.street FROM addresses WHERE addresses.user_id = users.user_id) s""")
def testin(self):
self.runtest(select([table], table.c.id.in_(1, 2, 3)),
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
