SQLAlchemy學(xué)習(xí)筆記（五）

當前表

my_db=# select * from users;
 id |  name  |    fullname    | password
----+--------+----------------+----------
  1 | ed     | Ed Jones       | 0000
  2 | wendy  | Wendy Williams | foobar
  3 | mary   | Mary Contrary  | xxf12
  4 | jack   | Jack Bean      | gjfdd
  5 | rickyy | ricky yu       | 1

my_db=# select * from addresses;
 id | email_address | user_id
----+---------------+---------
  1 | jack@qq.com   |       4
  2 | j12@gamil.com |       4
  3 | 1@qq.com      |       5
  4 | 2@qq.com      |       5
(4 行記錄)

刪除

讓我們刪除rickyy用戶，看看怎么回事，我們在會話中將該對象標記為已刪除，然后我們發(fā)出查詢，看看有沒有剩余的行；

>>> session.delete('rickyy')
>>> session.query(User).filter_by(name='rickyy').count()
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
WHERE users.id = %(param_1)s
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine {'param_1': 5}
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id
FROM addresses
WHERE %(param_1)s = addresses.user_id ORDER BY addresses.id
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine {'param_1': 5}
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine UPDATE addresses SET user_id=%(user_id)s WHERE addresses.id = %(addresses_id)s
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine ({'addresses_id': 3, 'user_id': None}, {'addresses_id': 4, 'user_id': None})
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine DELETE FROM users WHERE users.id = %(id)s
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine {'id': 5}
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine SELECT count(*) AS count_1
FROM (SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
WHERE users.name = %(name_1)s) AS anon_1
2017-03-11 09:17:32,045 INFO sqlalchemy.engine.base.Engine {'name_1': 'rickyy'}
0L

我們可以看到，用戶rickyy已經(jīng)被刪除了，我們再來看看Address表中的記錄是否還在

>>> session.query(Address).filter(Address.email_address.in_(['1@qq.com', '2@qq.com'])).count()
2017-03-11 09:21:54,078 INFO sqlalchemy.engine.base.Engine SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (%(email_address_1)s, %(email_address_2)s)) AS anon_1
2017-03-11 09:21:54,078 INFO sqlalchemy.engine.base.Engine {'email_address_2': '2@qq.com', 'email_address_1': '1@qq.com'}
2L

通過查詢，發(fā)現(xiàn)Address表中和用戶rikcyy關(guān)聯(lián)的地址仍然存在，我們提交改動試試。

>>> session.commit()
2017-03-11 09:22:39,924 INFO sqlalchemy.engine.base.Engine COMMIT
>>>

提交后查看一下我們的數(shù)據(jù)庫

my_db=# select * from addresses;
 id | email_address | user_id
----+---------------+---------
  1 | jack@qq.com   |       4
  2 | j12@gamil.com |       4
  3 | 1@qq.com      |
  4 | 2@qq.com      |
(4 行記錄)

我們可以看出，user_id列被置為NULL，但是行沒有被刪除，這是因為SQLAlchemy不會級聯(lián)刪除，我們必須告訴它這樣做。

配置刪除/級聯(lián)刪除

我們將在User.addresses關(guān)系上配置級聯(lián)選項以更改行為。雖然SQLAlchemy允許我們在任何時間點添加新的屬性和關(guān)系到映射，在這種情況下需要刪除現(xiàn)有關(guān)系，因此我們需要完全拆除映射并重新開始 - 我們將關(guān)閉會話；

>>> session.close()

使用declarative_base()新建一個基類

>>> Base = declarative_base()

下面我們將重新定義User類，并添加addresses關(guān)系，包括級聯(lián)配置。

>>> class User(Base):
...     __tablename__ = 'users'
...     id=Column(Integer, primary_key=True)
...     name=Column(String)
...     fullname=Column(String)
...     password=Column(String)
...     addresses = relationship('Address', back_populates="user",cascade='all,delete,delete-orphan')
...     def __repr__(self):
...             return "<User(name=%s, fullname=%s, password=%s)>" % (self.name, self.fullname, self.password)
...

重新定義Address類

>>> class Address(Base):
...     __tablename__ = 'addresses'
...     id = Column(Integer, primary_key=True)
...     email_address = Column(String, nullable=False)
...     user_id = Column(Integer, ForeignKey('users.id'))
...     user = relationship("User", back_populates="addresses")
...     def __repr__(self):
...             return "<Address(email_address=%s)>" % (self.email_address,)
...

接下來刪除用戶jack，可以從上面的表內(nèi)容看出我們的用戶jack有兩個地址。

# 通過主鍵查詢jack
>>> jack=session.query(User).get(4)
2017-03-11 09:42:16,760 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
WHERE users.id = %(param_1)s
2017-03-11 09:42:16,762 INFO sqlalchemy.engine.base.Engine {'param_1': 4}
>>> jack
<User(name=jack, fullname=Jack Bean, password=gjfdd)>

# 刪除jack的一個地址
>>> del jack.addresses[1]

# 查詢刪除后還有幾個地址
>>> session.query(Address).filter(Address.email_address.in_(['jack@qq.com','j12@gmail.com'])).count()
2017-03-11 09:43:46,957 INFO sqlalchemy.engine.base.Engine DELETE FROM addresses WHERE addresses.id = %(id)s
2017-03-11 09:43:46,957 INFO sqlalchemy.engine.base.Engine {'id': 2}
2017-03-11 09:43:47,000 INFO sqlalchemy.engine.base.Engine SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (%(email_address_1)s, %(email_address_2)s)) AS anon_1
2017-03-11 09:43:47,003 INFO sqlalchemy.engine.base.Engine {'email_address_2': 'j12@gmail.com', 'email_address_1': 'jack@qq.com'}
1L

我們刪除jack，和他關(guān)聯(lián)的所有地址將會被刪除：

>>> session.delete(jack)
>>> session.query(User).filter_by(name='jack').count()
2017-03-11 09:44:21,401 INFO sqlalchemy.engine.base.Engine DELETE FROM addresses WHERE addresses.id = %(id)s
2017-03-11 09:44:21,404 INFO sqlalchemy.engine.base.Engine {'id': 1}
2017-03-11 09:44:21,407 INFO sqlalchemy.engine.base.Engine DELETE FROM users WHERE users.id = %(id)s
2017-03-11 09:44:21,408 INFO sqlalchemy.engine.base.Engine {'id': 4}
2017-03-11 09:44:21,413 INFO sqlalchemy.engine.base.Engine SELECT count(*) AS count_1
FROM (SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
WHERE users.name = %(name_1)s) AS anon_1
2017-03-11 09:44:21,414 INFO sqlalchemy.engine.base.Engine {'name_1': 'jack'}
0L
>>> session.query(Address).filter(Address.email_address.in_(['jack@qq.com','j12@gmail.com'])).count()
2017-03-11 09:44:25,979 INFO sqlalchemy.engine.base.Engine SELECT count(*) AS count_1
FROM (SELECT addresses.id AS addresses_id, addresses.email_address AS addresses_email_address, addresses.user_id AS addresses_user_id
FROM addresses
WHERE addresses.email_address IN (%(email_address_1)s, %(email_address_2)s)) AS anon_1
2017-03-11 09:44:25,980 INFO sqlalchemy.engine.base.Engine {'email_address_2': 'j12@gmail.com', 'email_address_1': 'jack@qq.com'}
0L

最后我們提交到數(shù)據(jù)庫。

>>> session.commit()

再來看看當前的表內(nèi)容：

my_db=# select * from addresses;
 id | email_address | user_id
----+---------------+---------
  3 | 1@qq.com      |
  4 | 2@qq.com      |
(2 行記錄)

my_db=# select * from addresses;
 id | email_address | user_id
----+---------------+---------
  3 | 1@qq.com      |
  4 | 2@qq.com      |
(2 行記錄)

建立多對多的關(guān)系

在數(shù)據(jù)庫中建立n:m的關(guān)系，需要新增加一個關(guān)系表。
我們再定義一個BlogPost類，多對多關(guān)系我們要創(chuàng)建一個沒有映射的表作為關(guān)聯(lián)表：

>>> from sqlalchemy import Table, Text
>>> post_keywords = Table('post_keywords', Base.metadata,
...     Column('post_id', ForeignKey('posts.id'), primary_key=True),
...     Column('keyword_id', ForeignKey('keywords.id'), primary_key=True)
... )

我們上面創(chuàng)建了一個表，只有兩個字段分別參照兩個不同的表，posts keywords表。
接下來我們定義這兩個類

>>> 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,
...                             back_populates='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)
...     posts = relationship('BlogPost',
...                          secondary=post_keywords,
...                          back_populates='keywords')
...
...     def __init__(self, keyword):
...         self.keyword = keyword

因為發(fā)布文章表和關(guān)鍵詞表時n：m的關(guān)系，所以需要使用relationship()建立關(guān)系，這里使用了secondary來定義多對多的關(guān)系。
我們也希望我們的BlogPost類有一個作者字段。我們將它添加為另一個雙向關(guān)系，除了一個問題，我們將有一個單一的用戶可能有很多博客文章。當我們訪問User.posts時，我們希望能夠進一步過濾結(jié)果，以便不加載整個集合。為此，我們使用一個被relationship（）稱為lazy ='dynamic'的設(shè)置，該屬性在屬性上配置一個備用加載器策略。

>>> BlogPost.author = relationship(User, back_populates="posts")
>>> User.posts = relationship(BlogPost, back_populates="author", lazy="dynamic")

創(chuàng)建一個新表

>>> Base.metadata.create_all(engine)

現(xiàn)在我們發(fā)布博客

>>> wendy = session.query(User).\
...                 filter_by(name='wendy').\
...                 one()
>>> post = BlogPost("Wendy's Blog Post", "This is a test", wendy)
>>> session.add(post)

我們將關(guān)鍵字存儲在數(shù)據(jù)庫中，但是現(xiàn)在還沒有，所以我們可以創(chuàng)建它們：

>>> post.keywords.append(Keyword('wendy'))
>>> post.keywords.append(Keyword('firstpost')

我們現(xiàn)在可以使用關(guān)鍵字firstpost查找所有博文。我們將使用any運算符來定位“任何關(guān)鍵字具有關(guān)鍵字字符串firstpost的博客帖子

>>> session.query(BlogPost).filter(BlogPost.keywords.any(keyword='firstpost')).all()
2017-03-11 10:23:44,809 INFO sqlalchemy.engine.base.Engine INSERT INTO keywords (keyword) VALUES (%(keyword)s) RETURNING keywords.id
2017-03-11 10:23:44,809 INFO sqlalchemy.engine.base.Engine {'keyword': 'wendy'}
2017-03-11 10:23:44,809 INFO sqlalchemy.engine.base.Engine INSERT INTO keywords (keyword) VALUES (%(keyword)s) RETURNING keywords.id
2017-03-11 10:23:44,809 INFO sqlalchemy.engine.base.Engine {'keyword': 'firstpost'}
2017-03-11 10:23:44,809 INFO sqlalchemy.engine.base.Engine INSERT INTO posts (user_id, headline, body) VALUES (%(user_id)s, %(headline)s, %(body)s) RETURNING posts.id
2017-03-11 10:23:44,825 INFO sqlalchemy.engine.base.Engine {'body': 'This is a test', 'headline': "Wendy's Blog Post", 'user_id': 2}
2017-03-11 10:23:44,857 INFO sqlalchemy.engine.base.Engine INSERT INTO post_keywords (post_id, keyword_id) VALUES (%(post_id)s, %(keyword_id)s)
2017-03-11 10:23:44,857 INFO sqlalchemy.engine.base.Engine ({'keyword_id': 2, 'post_id': 1}, {'keyword_id': 1, 'post_id': 1})
2017-03-11 10:23:44,872 INFO sqlalchemy.engine.base.Engine SELECT posts.id AS posts_id, posts.user_id AS posts_user_id, posts.headline AS posts_headline, posts.body AS posts_body
FROM posts
WHERE EXISTS (SELECT 1
FROM post_keywords, keywords
WHERE posts.id = post_keywords.post_id AND keywords.id = post_keywords.keyword_id AND keywords.keyword = %(keyword_1)s)
2017-03-11 10:23:44,872 INFO sqlalchemy.engine.base.Engine {'keyword_1': 'firstpost'}
[BlogPost("Wendy's Blog Post", 'This is a test', <User(name=wendy, fullname=Wendy Williams, password=foobar)>)]

補充上面關(guān)系E-R圖

E-R圖

參考官方文檔：http://docs.sqlalchemy.org/en/latest/orm/tutorial.html#building-a-relationship

最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容