forked from ghosert/VimProject
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_sqlalchemy.py
More file actions
109 lines (77 loc) · 3.68 KB
/
Copy pathtest_sqlalchemy.py
File metadata and controls
109 lines (77 loc) · 3.68 KB
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
from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:', echo=True)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
password = Column(String)
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)
# users_table = User.__table__
metadata = Base.metadata
metadata.create_all(engine)
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship, backref
Session = sessionmaker(bind=engine)
session = Session()
ed_user = User('ed', 'Ed Jones', 'edspassword')
session.add(ed_user)
session.add_all([User('wendy', 'Wendy Williams', 'foobar'), User('mary', 'Mary Contrary', 'xxg527'), User('fred', 'Fred Flinstone', 'blah')])
print session.query(User).all()
from sqlalchemy import Text
# association table
post_keywords = Table('post_keywords', metadata,
Column('post_id', Integer, ForeignKey('posts.id')),
Column('keyword_id', Integer, ForeignKey('keywords.id'))
)
class BlogPost(Base):
__tablename__ = 'posts'
id = Column(Integer, primary_key=True)
user_id = Column(Integer, ForeignKey('users.id'))
headline = Column(String(255), nullable=False)
body = Column(Text)
# many to many BlogPost<->Keyword
keywords = relationship('Keyword', secondary=post_keywords, backref='posts')
def __init__(self, headline, body, author):
self.author = author
self.headline = headline
self.body = body
def __repr__(self):
return "BlogPost(%r, %r, %r)" % (self.headline, self.body, self.author)
class Keyword(Base):
__tablename__ = 'keywords'
id = Column(Integer, primary_key=True)
keyword = Column(String(50), nullable=False, unique=True)
def __init__(self, keyword):
self.keyword = keyword
# "dynamic" loading relationship to User, and notice we don't define posts/relationship() in User class.
BlogPost.author = relationship(User, backref=backref('posts', lazy='dynamic'))
metadata.create_all(engine)
wendy = session.query(User).filter_by(name='wendy').one()
post = BlogPost("Wendy's Blog Post", "This is a test", wendy)
session.add(post)
post.keywords.append(Keyword('wendy'))
post.keywords.append(Keyword('firstpost'))
session.commit()
print '============================Two SQLs produced without contains_eager============================================='
from sqlalchemy.orm import contains_eager
blogpost = session.query(BlogPost).filter(BlogPost.keywords.any(keyword='firstpost')).all()
print blogpost[0].keywords[1].keyword
session.commit()
print '============================One SQL produced wit contains_eager, it means contains_eager works for both JOIN and EXISTS sql=================================================='
blogpost = session.query(BlogPost).filter(BlogPost.keywords.any(keyword='firstpost')).options(contains_eager(BlogPost.keywords)).all()
print blogpost[0].keywords[1].keyword
print '================================================================================================================='
blog = session.query(BlogPost).filter(BlogPost.author==wendy).filter(BlogPost.keywords.any(keyword='firstpost')).all()
print blog[0].keywords[0].posts
# Even if we fail to define posts/relationship() in User class, we can still use it like below:
wendy = session.query(User).filter_by(name='wendy').one()
print wendy.posts[0].headline