Copyright 2008 Vladimir Magamedov import sqlalchemy print Using sqlalc

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
# Copyright (c) 2008 Vladimir Magamedov
import sqlalchemy
print 'Using sqlalchemy v%s' % sqlalchemy.__version__, '\n'+'-'*50
from sqlalchemy import create_engine
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
from sqlalchemy.orm import mapper
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relation, backref
from sqlalchemy.orm import aliased
from sqlalchemy import and_, or_
from sqlalchemy.orm import join
print '#-------------------------------------------- Create DataBase'
engine = create_engine('sqlite:///:memory:')#, echo=True)
metadata = MetaData()
users_table = Table('users', metadata,
Column('id', Integer, primary_key=True),
Column('name', String(40)),
Column('fullname', String(100)),
Column('password', String(15)),
)
addresses_table = Table('addresses', metadata,
Column('id', Integer, primary_key=True),
Column('email_address', String(100), nullable=False),
Column('user_id', Integer, ForeignKey('users.id'))
)
#user = relation('users', backref=backref('addresses', order_by='id'))
metadata.create_all(engine)
print '#-------------------------------------------- Mapping'
class User(object):
def __init__(self, name, fullname, password):
self.name = name
self.fullname = fullname
self.password = password
def __repr__(self):
return "<User('%s','%s', '%s')>" % \
(self.name, self.fullname, self.password)
class Address(object):
def __init__(self, email_address):
self.email_address = email_address
def __repr__(self):
return "<Address('%s')>" % self.email_address
mapper(User, users_table,
properties={'addresses':relation(Address, backref='user')})
mapper(Address, addresses_table)
print '#-------------------------------------------- Create Session'
#print help(sessionmaker), '\n'+'-'*50
Session = sessionmaker(bind=engine)#autoflush=True, autocommit=True)
#Session.configure(bind=engine)
session = Session()
print '#-------------------------------------------- Fill data to DB'
ed_user = User('ed', 'Ed Jones', 'edspassword')
print str(ed_user.id)
session.add(ed_user)
session.commit()
user = session.query(User).filter_by(name='ed').first()
print user, ed_user is user
session.add(User('wendy', 'Wendy Williams', 'foobar'))
session.add(User('mary', 'Mary Contrary', 'xxg527'))
session.add(User('fred', 'Fred Flinstone', 'blah'))
ed_user.password = 'f8s7ccs'
print 'dirty records:', session.dirty
print 'new records:', session.new
session.commit()
print ed_user.addresses
ed_user.addresses += [Address(email_address='ed@google.com'),
Address(email_address='edward@google.com')]
session.commit()
print ed_user.addresses
print ed_user.addresses[1].user
print '#-------------------------------------------- Query data in DB'
print '#get all records'
for user in session.query(User).order_by(User.id):
print user.name
else: print
print '#get all records in named tuples'
for row in session.query(User, User.name).all():
print row.User, row.name
else: print
print '#get all records with control names'
user_alias = aliased(User, name='user_alias')
for row in session.query(user_alias, user_alias.name.label('name_label')).all():
print row.user_alias, row.name_label
else: print
print '# limit and offset'
print list(session.query(User).order_by(User.id)[1:3])
print '# with filter'
q = session.query(User.name, User.fullname)
print q.filter_by(fullname = 'Ed Jones').all()
print q.filter(User.fullname == 'Ed Jones').all()
print q.filter(User.name != 'ed').all()
print q.filter(User.name.like('%ed%')).all()
print q.filter(User.name == None).all()
print q.filter(and_(User.name.like('%ed%'), User.fullname == 'Ed Jones')).all()
#print q.filter(User.name.match('wendy')).all()
print session.query(User).filter('id<:value and name=:name').\
params(value=224, name='fred').order_by(User.id).one()
#for user in session.query(User).filter(users_table.c.name=='ed'):
# print user.name, user.fullname
print '# filter with joins'
print session.query(User, Address).filter(User.id==Address.user_id).\
filter(Address.email_address=='ed@google.com').all()
print session.query(User, Address).select_from(join(User, Address)).\
filter(Address.email_address=='ed@google.com').all()
print session.query(User, Address).select_from(join(User, Address, User.id==Address.user_id)).\
filter(Address.email_address=='ed@google.com').all()
print session.query(User, Address).join(User.addresses).\
filter(Address.email_address=='ed@google.com').all()
print '# aliased tables'
adalias1 = aliased(Address)
adalias2 = aliased(Address)
q = session.query(User.name, adalias1.email_address, adalias2.email_address)
q = q.join((adalias1, User.addresses), (adalias2, User.addresses))
q = q.filter(adalias1.email_address=='ed@google.com')
q = q.filter(adalias2.email_address=='edward@google.com')
print q.all()
print '#-------------------------------------------- End'