SQLAlchemy模型使用
2016-08-03 00:20
274 查看
SQLAchemy模型使用
简介：
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
# 条件
ret = session.query(Users).filter_by(name='alex').all()
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
from sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(
or_(
Users.id < 2,
and_(Users.name == 'eric', Users.id > 3),
Users.extra != ""
)).all()
# 通配符
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()
# 限制
ret = session.query(Users)[1:2]
# 排序
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()
# 分组
from sqlalchemy.sql import func
ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)).group_by(Users.name).all()
ret = session.query(
func.max(Users.id),
func.sum(Users.id),
func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()
# 连表
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
ret = session.query(Person).join(Favor).all()
ret = session.query(Person).join(Favor, isouter=True).all()
# 组合
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()
View Code
4. SQLAlchemy外键和关系
4.1. 外键约束
使用 ForeignKey 来定义一个外键约定:
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, CHAR,BIGINT,ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://mysql:123456@10.0.0.8:3306/test", max_overflow=5) #创建数据库连接
DBSession=sessionmaker(engine)#创建了一个自定义了的 Session类
session=DBSession()
BaseModel = declarative_base()#创建对象的基类
class Blog(BaseModel):
__tablename__='blog'
id=Column(BIGINT,primary_key=True,autoincrement=True)
title=Column(String(64),server_default='',nullable=False)
text=Column(String(256),server_default='',nullable=False)
user=Column(BIGINT,ForeignKey('user.id'),index=True,nullable=False)#将user表中id字段作为外键，使用ForeignKey指定
class User(BaseModel):
__tablename__='user'
id=Column(BIGINT,primary_key=True,autoincrement=True)
name=Column(String(32),server_default='',nullable=False)
username=Column(String(32),index=True,server_default='',nullable=True)
passwd=Column(String(64),server_default='',nullable=False)
创建时:
BaseModel.metadata.create_all(engine) user=User(name='first',username='pan',passwd='123456') session.add(user) session.flush() blog = Blog(title='frist', user=user.id) session.add(blog) session.commit()
session.flush()是进行数据库交互, 但是事务并没有提交. 进行数据库交互之后,
user.id才有值.
定义了外键, 对查询来说, 并没有影响. 外键只是单纯的一条约束而已. 当然, 可以在外键上定义一些关联的事件操作, 比如当外键条目被删除时, 字段置成
null, 或者关联条目也被删除等.
4.2. 关系定义
要定义关系, 必有使用 ForeignKey 约束. 当然, 这里说的只是在定义模型时必有要有, 至于数据库中是否真有外键约定, 这并不重要.
接下来我们来了解几个关于外键(Foreign Key)的小知识：
1. FOREIGN KEY 约束是大多数(但不是所有)的关系型数据库中可以链接到主键列，或者拥有UNIQUE约束的列。
2. FOREIGN KEY 能够引用多重列主键，并且其自身拥有多重列，被称为“复合外键”(composite foreign key)。其也能够引用这些列的子集(subset)。(注：这地方不太明白)
3. FOREIGN KEY 列作为对于其引用的列或者行的变化的响应能够自动更新其自身，比如CASCADE引用操作，这些都是内置于关系型数据库的功能之一。
4. FOREIGN KEY 能够引用其自身的表，这个就涉及到“自引用”(self-referential)的外键了。
5. 更多关于外键的资料可以参考Foreign Key – Wikipedia。
#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, CHAR,BIGINT,ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
engine = create_engine("mysql+pymysql://mysql:123456@10.0.0.8:3306/test", max_overflow=5) #创建数据库连接
DBSession=sessionmaker(engine)#创建了一个自定义了的 Session类
session=DBSession()
BaseModel = declarative_base()#创建对象的基类
class Blog(BaseModel):
__tablename__='blog'
id=Column(BIGINT,primary_key=True,autoincrement=True)
title=Column(String(64),server_default='',nullable=False)
text=Column(String(256),server_default='',nullable=False)
user=Column(BIGINT,ForeignKey('user.id'),index=True,nullable=False)
user_obj=relationship('User')
class User(BaseModel):
__tablename__='user'
id=Column(BIGINT,primary_key=True,autoincrement=True)
name=Column(String(32),server_default='',nullable=False)
username=Column(String(32),index=True,server_default='',nullable=True)
passwd=Column(String(64),server_default='',nullable=False)
blog_list=relationship('Blog',backref='Blog.user')
#relationship函数是sqlalchemy对关系之间提供的一种便利的调用方式, backref参数则对关系提供反向引用的声明。
#大致原理应该就是sqlalchemy在运行时对Blog对象动态的设置了一个指向所属User对象的属性，这样就能在实际开发中使逻辑关系更加清晰，代码更加简洁了。
BaseModel.metadata.create_all(engine)
user=User(name='first',username='pan',passwd='123456')
session.add(user)
session.flush()
blog = Blog(title='frist', user=user.id)
session.add(blog)
session.commit()
关系只是 SQLAlchemy 提供的工具, 与数据库无关, 所以任何时候添加都是可以的.
上面的 User-Blog 是一个"一对多"关系, 通过
Blog的
user这个 ForeignKey , SQLAlchemy 可以自动处理关系的定义. 在查询时, 返回的结果自然也是, 一个是列表, 一个是单个对象:
ret=session.query(Blog).get(1).user_obj ret1=session.query(User).get(1).blog_list print(ret.passwd,ret1[0].id)#ret.passwd=User.passwd ret1[0].id=Blog.user列表中第一个值
#blog_list=relationship('Blog',backref='Blog.user')
#user_obj=relationship('User')
#relationship指定对象不同引用方法不同
这种关系的定义, 并不影响查询并获取对象的行为, 不会添加额外的
join操作. 在对象上取一个
user_obj或者取
blog_list都是发生了一个新的查询操作.
上面的关系定义, 对应的属性是实际查询出的实例列表, 当条目数多的时候, 这样可能会有问题. 比如用户名下有成千上万的文章, 一次全取出就太暴力了. 关系对应的属性可以定义成一个 Query :
class User(BaseModel):
__tablename__ = 'user'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(32), server_default='', nullable=False)
blog_list = relationship('Blog', order_by='Blog.user', lazy="dynamic")
这样在获取实例时就可以自由控制了:
session.query(User).get(1).blog_list.all() session.query(User).get(1).blog_list.filter(Blog.title == '1').first()
4.3. 关系的查询
关系定义之后, 除了在查询时会有自动关联的效果, 在作查询时, 也可以对定义的关系做操作:
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(Integer, autoincrement=True, primary_key=True)
title = Column(Unicode(32), server_default='')
user = Column(Integer, ForeignKey('user.id'), index=True)
user_obj = relationship('User')
class User(BaseModel):
__tablename__ = 'user'
id = Column(Integer, autoincrement=True, primary_key=True)
name = Column(Unicode(32), server_default='')
blogs = relationship('Blog')
对于 一对多 的关系, 使用
any()函数查询:
user = session.query(User).filter(User.blogs.any(Blog.title == 'first')).first()
SQLAlchemy 会使用
exists条件, 类似于:
SELECT *FROM user WHERE EXISTS (SELECT 1 FROM blog WHERE user.id = blog.user AND blog.title = ?) LIMIT ? OFFSET ?
反之, 如果是 多对一 的关系, 则使用
has()函数查询:
blog = session.query(Blog).filter(Blog.user_obj.has(User.name == 'pan')).first()
最后的 SQL 语句都是一样的.
4.4. 关系的获取形式
前面介绍的关系定义中, 提到了两种关系的获取形式, 一种是:
user_obj = relationship('User')
这种是在对象上获取关系对象时, 再去查询.
另一种是:
blog_list = relationship('Blog', lazy="dynamic")
这种的结果, 是在对象上获取关系对象时, 只返回 Query , 而查询的细节由人为来控制.
总的来说, 关系的获取分成两种, Lazy 或 Eager . 在直接查询层面, 上面两种都属于 Lazy 的方式, 而 Eager 的一种, 就是在获取对象时的查询语句, 是直接带
join的, 这样关系对象的数据在一个查询语句中就直接获取到了:
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(BIGINT, primary_key=True, autoincrement=True)
title = Column(String(64), server_default='', nullable=False)
user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
user_obj = relationship('User', lazy='joined', cascade='all')
class User(BaseModel):
__tablename__ = 'user'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(32), server_default='', nullable=False)
这样在查询时:
blog = session.query(Blog).first() print( blog.user_obj)
便会多出
LEFT OUTER JOIN的语句, 结果中直接获取到对应的
User实例对象.
也可以把
joined换成子查询,
subquery:
class User(BaseModel):
__tablename__ = 'user'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(32), server_default='', nullable=False)
blog_list = relationship('Blog', cascade='all', lazy='subquery')
if __name__ == '__main__':
session = Session()
user = session.query(User).first()
session.commit()
子查询会用到临时表.
上面定义的:
blog_list = relationship('Blog', lazy="dynamic")
user_obj = relationship('User', lazy='joined')
blog_list = relationship('Blog', lazy='subquery')
都算是一种默认方式. 在具体使用查询时, 还可以通过
options()方法定义关联的获取方式:
from sqlalchemy.orm import lazyload, joinedload, subqueryload
user = session.query(User).options(lazyload('blog_list')).first()
print (user.blog_list)
更多的用法:
session.query(Parent).options(
joinedload('foo').joinedload('bar').joinedload('bat')
).all()
session.query(A).options(
defaultload("atob").joinedload("btoc")
).all()
session.query(MyClass).options(lazyload('*'))
session.query(MyClass).options(
lazyload('*'), joinedload(MyClass.widget)
)
session.query(User, Address).options(Load(Address).lazyload('*'))
如果关联的定义之前是 Lazy 的, 但是实际使用中, 希望在手工 join 之后, 把关联对象直接包含进结果实例, 可以使用
contains_eager()来包装一下:
from sqlalchemy.orm import contains_eager blog = session.query(Blog).join(Blog.user_obj)\ .options(contains_eager(Blog.user_obj)).first() print blog.user_obj
4.5. 关系的表现形式
关系在对象属性中的表现, 默认是列表, 但是, 这不是唯一的形式. 根据需要, 可以作成 dictionary , set 或者其它你需要的对象.
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(Integer, autoincrement=True, primary_key=True)
title = Column(Unicode(32), server_default='')
user = Column(Integer, ForeignKey('user.id'), index=True)
user_obj = relationship('User')
class User(BaseModel):
__tablename__ = 'user'
id = Column(Integer, autoincrement=True, primary_key=True)
name = Column(Unicode(32), server_default='')
blogs = relationship('Blog')
对于上面的两个模型:
user = session.query(User).first() print (user.blogs)
现在
user.blogs是一个列表. 我们可以在
relationship()调用时通过
collection_class参数指定一个类, 来重新定义关系的表现形式:
user = User(name='XXX') session.add_all([Blog(title='A', user_obj=user), Blog(title='B', user_obj=user)]) session.commit() user = session.query(User).first() print (user.blogs)
set , 集合:
blogs = relationship('Blog', collection_class=set)
#InstrumentedSet([<__main__.Blog object at 0x1a58710>, <__main__.Blog object at 0x1a587d0>])
attribute_mapped_collection , 字典, 键值从属性取:
from sqlalchemy.orm.collections import attribute_mapped_collection
blogs = relationship('Blog', collection_class=attribute_mapped_collection('title'))
#{'A': <__main__.Blog object at 0x20ed810>, 'B': <__main__.Blog object at 0x20ed8d0>}
- 如果
title
重复的话, 结果会覆盖.
- mapped_collection , 字典, 键值自定义:
from sqlalchemy.orm.collections import mapped_collection
blogs = relationship('Blog', collection_class=mapped_collection(lambda blog: blog.title.lower()))
4.6. 多对多关系
先考虑典型的多对多关系结构:
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(BIGINT, primary_key=True, autoincrement=True)
title = Column(String(64), server_default='', nullable=False)
tag_list = relationship('Tag')
tag_list = relationship('BlogAndTag')
class Tag(BaseModel):
__tablename__ = 'tag'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(16), server_default='', nullable=False)
class BlogAndTag(BaseModel):
__tablename__ = 'blog_and_tag'
id = Column(BIGINT, primary_key=True, autoincrement=True)
blog = Column(BIGINT, ForeignKey('blog.id'), index=True)
tag = Column(BIGINT, ForeignKey('tag.id'), index=True)
create = Column(BIGINT, index=True, server_default='0')
在
Blog中的:
tag_list = relationship('Tag')
显示是错误的, 因为在
Tag中并没有外键. 而:
tag_list = relationship('BlogAndTag')
这样虽然正确, 但是
tag_list的关系只是到达
BlogAndTag这一层, 并没有到达我们需要的
Tag.
这种情况下, 一个多对多关系是有三张表来表示的, 在定义 relationship 时, 就需要一个
secondary参数来指明关系表:
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(BIGINT, primary_key=True, autoincrement=True)
title = Column(String(64), server_default='', nullable=False)
tag_list = relationship('Tag', secondary=lambda: BlogAndTag.__table__)
#是用lambda可以使后面跟的类无需提前定义，如果直接secondary=classname，则class需天气定义
class Tag(BaseModel):
__tablename__ = 'tag'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(16), server_default='', nullable=False)
class BlogAndTag(BaseModel):
__tablename__ = 'blog_and_tag'
id = Column(BIGINT, primary_key=True, autoincrement=True)
blog = Column(BIGINT, ForeignKey('blog.id'), index=True)
tag = Column(BIGINT, ForeignKey('tag.id'), index=True)
create = Column(BIGINT, index=True, server_default='0')
这样, 在操作时可以直接获取到对应的实例列表:
blog = session.query(Blog).filter(Blog.title == 'a').one() print (blog.tag_list)
访问
tag_list时, SQLAlchemy 做的是一个普通的多表查询.
tag_list属性同时支持赋值操作:
session = Session() blog = session.query(Blog).filter(Blog.title == 'a').one() blog.tag_list = [Tag(name='t1')] session.commit()
提交时, SQLAlchemy 总是会创建
Tag, 及对应的关系
BlogAndTag.
而如果是:
session = Session() blog = session.query(Blog).filter(Blog.title == 'a').one() blog.tag_list = [] session.commit() tag = session.query(Tag).filter(Tag.name == 'x').one() blog.tag_list.remove(tag) session.commit()
那么 SQLAlchemy 只会删除对应的关系
BlogAndTag, 不会删除实体
Tag.
如果你直接删除实体, 那么对应的关系是不会自动删除的:
session = Session() blog = session.query(Blog).filter(Blog.title == 'a').one() tag = Tag(name='ok') blog.tag_list = [tag] session.commit() tag = session.query(Tag).filter(Tag.name == 'ok').one() session.delete(tag) session.commit()
4.7. Cascades 自动关系处理
前面提到的, 当操作关系, 实体时, 与其相关联的关系, 实体是否会被自动处理的问题, 在 SQLAlchemy 中是通过 Cascades 机制来定义和解决的. ( Cascades 这个词是来源于Hibernate .)
cascade是一个
relationship的参数, 其值是逗号分割的多个字符串, 以表示不同的行为. 默认值是 " save-update, merge" , 稍后会介绍每个词项的作用.
这里的所有规则介绍, 只涉及从 Parent 到 Child , Parent 即定义 relationship的类. 不涉及backref .
cascade所有的可选字符串项是:
- all , 所有操作都会自动处理到关联对象上.
- save-update , 关联对象自动添加到会话.
- delete , 关联对象自动从会话中删除.
- delete-orphan , 属性中去掉关联对象, 则会话中会自动删除关联对象.
- merge ,
session.merge()
时会处理关联对象. - refresh-expire ,
session.expire()
时会处理关联对象. - expunge ,
session.expunge()
时会处理关联对象.
- save-update
- 当一个对象被添加进 session 后, 此对象标记为 save-update 的
relationship
关系对象也会同时添加进这个 session .
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
7
8
9 class User(BaseModel):
10 __tablename__ = 'user'
11
12 id = Column(BIGINT, primary_key=True, autoincrement=True)
13 name = Column(String(32), server_default='', nullable=False)
14
15 blog_list = relationship('Blog', cascade='')
16 blog_list_auto = relationship('Blog', cascade='save-update')
17
18
19 if __name__ == '__main__':
20
21 session = Session()
22
23 user = User(name=u'哈哈')
24 blog = Blog(title=u'第一个')
25 user.blog_list = [blog]
26 #user.blog_list_auto = [blog]
27 session.add(user)
28 print (blog in session)
29 session.commit()
- delete
- 当一个对象在 session 中被标记为删除时, 其属性中 relationship 关联的对象也会被标记成删除, 否则, 关联对象中的对应外键字段会被改成
NULL
, 不能为NULL
则报错.
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
7
8
9 class User(BaseModel):
10 __tablename__ = 'user'
11
12 id = Column(BIGINT, primary_key=True, autoincrement=True)
13 name = Column(String(32), server_default='', nullable=False)
14
15 blog_list = relationship('Blog', cascade='save-update, delete')
16
17
18 if __name__ == '__main__':
19 session = Session()
20
21 #user = User(name=u'用户')
22 #user.blog_list = [Blog(title=u'哈哈')]
23 #session.add(user)
24 user = session.query(User).first()
25 session.delete(user)
26 session.commit()
- delete-orphan
- 当 relationship 属性变化时, 被 "去掉" 的对象会被自动删除. 比如之前是:
user.blog_list = [blog, blog2]
- 现在变成:
user.blog_list = [blog2]
- 那么
blog
这个关联实体是会自动删除的. 这各机制只适用于 "一对多" 的关系中, "多对多" 和反过来的 "多对一" 都不适用. 在relationship
定义时, 可以添加single_parent = True
参数来强制约束. 当然, 在实现上 SQLAlchemy 是会先查出所有关联实体, 然后计算差集确认哪些需要被删除.
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
7
8
9 class User(BaseModel):
10 __tablename__ = 'user'
11
12 id = Column(BIGINT, primary_key=True, autoincrement=True)
13 name = Column(String(32), server_default='', nullable=False)
14
15 blog_list = relationship('Blog', cascade='save-update, delete-orphan')
16
17
18 if __name__ == '__main__':
19
20 session = Session()
21
22 #user = User(name=u'用户')
23 #blog = Blog(title=u'一')
24 #blog2 = Blog(title=u'二')
25 #user.blog_list = [blog, blog2]
26 #session.add(user)
27 user = session.query(User).first()
28 blog2 = session.query(Blog).filter(Blog.title == u'二').first()
29 user.blog_list = [blog2]
30 #session.delete(user)
31 session.commit()
- merge
- 这个选项是标识在
session.merge()
时处理关联对象.session.merge()
的作用, 是把一个会话外的实例, "整合"进会话, 比如 "有则修改, 无则创建" 就是典型的一种 "整合":
1 user = User(id=1, name="1") 2 session.add(user) 3 session.commit() 4 5 user = User(id=1) 6 user = session.merge(user) 7 print user.name 8 9 user = User(id=1, name="2") 10 user = session.merge(user) 11 session.commit()
cascade
中的merge
作用:
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(BIGINT, primary_key=True, autoincrement=True)
5 title = Column(String(64), server_default='', nullable=False)
6 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
7
8
9 class User(BaseModel):
10 __tablename__ = 'user'
11
12 id = Column(BIGINT, primary_key=True, autoincrement=True)
13 name = Column(String(32), server_default='', nullable=False)
14
15 blog_list = relationship('Blog',
16 cascade='save-update, delete, delete-orphan, merge')
17
18
19 if __name__ == '__main__':
20
21 session = Session()
22
23 user = User(id=1, name='1')
24 session.add(user)
25 session.commit(user)
26
27 user = User(id=1, blog_list=[Blog(title='哈哈')])
28 session.merge(user)
29
30 session.commit()
- refresh-expire
1 当使用 session.expire() 标识一个对象过期时, 此对象的关联对象是否也被标识为过期(访问属性会重新查询数据库).
2
3 class Blog(BaseModel):
4 __tablename__ = 'blog'
5
6 id = Column(BIGINT, primary_key=True, autoincrement=True)
7 title = Column(String(64), server_default='', nullable=False)
8 user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
9
10
11 class User(BaseModel):
12 __tablename__ = 'user'
13
14 id = Column(BIGINT, primary_key=True, autoincrement=True)
15 name = Column(String(32), server_default='', nullable=False)
16
17 blog_list = relationship('Blog',
18 cascade='save-update, delete, delete-orphan, merge, refresh-expire')
19
20
21 if __name__ == '__main__':
22
23 session = Session()
24
25 #user = User(id=1, name='1')
26 #blog = Blog(title="abc")
27 #user.blog_list = [blog]
28 #session.add(user)
29
30 user = session.query(User).first()
31 blog = user.blog_list[0]
32 print (user.name)
33 print (blog.title)
34 session.expire(user)
35 print ('EXPIRE')
36 print (user.name)
37 print (blog.title)
38
39 session.commit()
- expunge
- 与 merge 相反, 当 session.expunge() 把对象从会话中去除的时候, 此对象的关联对象也同时从会话中消失.
class Blog(BaseModel):
__tablename__ = 'blog'
id = Column(BIGINT, primary_key=True, autoincrement=True)
title = Column(String(64), server_default='', nullable=False)
user = Column(BIGINT, ForeignKey('user.id'), index=True, nullable=False)
class User(BaseModel):
__tablename__ = 'user'
id = Column(BIGINT, primary_key=True, autoincrement=True)
name = Column(String(32), server_default='', nullable=False)
blog_list = relationship('Blog', cascade='delete, delete-orphan, expunge')
if __name__ == '__main__':
session = Session()
user = User(name=u'用户')
blog = Blog(title=u'第一个')
user.blog_list = [blog]
session.add(user)
session.add(blog)
session.expunge(user)
print (blog in session)
#session.commit()
- 4.8. 属性代理
- 考虑这样的情况, 关系是关联的整个模型对象的, 但是, 有时我们对于这个关系, 并不关心整个对象, 只关心其中的某个属性. 考虑下面的场景:
1 from sqlalchemy.ext.associationproxy import association_proxy
2
3 class Blog(BaseModel):
4 __tablename__ = 'blog'
5
6 id = Column(Integer, autoincrement=True, primary_key=True)
7 title = Column(Unicode(32), nullable=False, server_default='')
8 user = Column(Integer, ForeignKey('user.id'), index=True)
9
10
11 class User(BaseModel):
12 __tablename__ = 'user'
13
14 id = Column(Integer, autoincrement=True, primary_key=True)
15 name = Column(Unicode(32), nullable=False, server_default='')
16
17 blog_list = relationship('Blog')
18 blog_title_list = association_proxy('blog_list', 'title')
blog_list
是一个正确的一对多关系. 下面的blog_title_list
就是这个关系上的一个属性代理.blog_title_list
只处理blog_list
这个关系中对应的对象的title
属性, 包括获取和设置两个方向.
1 session = Session() 2 3 user = User(name='xxx') 4 user.blog_list = [Blog(title='ABC')] 5 session.add(user) 6 session.commit() 7 8 user = session.query(User).first() 9 print (user.blog_title_list)
- 上面是获取属性的示例. 在"设置", 或者说"创建"时, 直接操作是有错的:
1 user = session.query(User).first() 2 user.blog_title_list = ['NEW'] 3 session.add(user) 4 session.commit()
- 原因在于, 对于类
Blog
的初始化形式.association_proxy('blog_list', 'title')中的title
只是获取时的属性定义, 而在上面的设置过程中, 实际上的调用形式为:
Blog('NEW')
Blog
类没有明确定义__init__()
方法, 所有这种形式的调用会报错. 可以把__init__()
方法补上:- 这样调用就没有问题了.
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(Integer, autoincrement=True, primary_key=True)
5 title = Column(Unicode(32), nullable=False, server_default='')
6 user = Column(Integer, ForeignKey('user.id'), index=True)
7
8 def __init__(self, title):
9 self.title = title
- 另一个方法, 是在调用
association_proxy()
时使用creator
参数明确定义"值"和"实例"的关系:
1 class User(BaseModel):
2 __tablename__ = 'user'
3
4 id = Column(Integer, autoincrement=True, primary_key=True)
5 name = Column(Unicode(32), nullable=False, server_default='')
6
7 blog_list = relationship('Blog')
8 blog_title_list = association_proxy('blog_list', 'title',
9 creator=lambda t: User(title=t))
creator
定义的方法, 返回的对象可以被对应的blog_list
关系接收即可.- 在查询方面, 多对一 的关系代理上, 可以直接使用属性:、
1 class Blog(BaseModel):
2 __tablename__ = 'blog'
3
4 id = Column(Integer, autoincrement=True, primary_key=True)
5 title = Column(Unicode(32), server_default='')
6 user = Column(Integer, ForeignKey('user.id'), index=True)
7
8 user_obj = relationship('User')
9 user_name = association_proxy('user_obj', 'name')
- 查询:
blog = session.query(Blog).filter(Blog.user_name == 'XX').first()
- 反过来的 一对多 关系代理上, 可以使用
contains()
函数:
user = session.query(User).filter(User.blogs_title.contains('A')).first()
5. SQLAlchemy会话与事务控制
5.1. 基本使用
SQLAlchemy 的 session 是用于管理数据库操作的一个像容器一样的东西. 模型实例对象本身独立存在, 而要让其修改(创建)生效, 则需要把它们加入某个 session . 同时你也可以把模型实例对象从 session 中去除. 被 session 管理的实例对象, 在
session.commit()时被提交到数据库. 同时
session.rollback()是回滚变更.
session.flush()的作用是在事务管理内与数据库发生交互, 对应的实例状态被反映到数据库. 比如自增 ID 被填充上值.
user = User(name='名字') session.add(user) session.commit()
try: user = session.Query(User).first() user.name = u'改名字 session.commit() except: session.rollback()
5.2. for update
SQLAlchemy 的 Query 支持
select ... for update / share.
session.Query(User).with_for_update().first() session.Query(User).with_for_update(read=True).first()
完整形式是:
with_for_update(read=False, nowait=False, of=None)
- read
- 是标识加互斥锁还是共享锁. 当为
True
时, 即for share
的语句, 是共享锁. 多个事务可以获取共享锁, 互斥锁只能一个事务获取. 有"多个地方"都希望是"这段时间我获取的数据不能被修改, 我也不会改", 那么只能使用共享锁. - nowait
- 其它事务碰到锁, 是否不等待直接"报错".
- of
- 指明上锁的表, 如果不指明, 则查询中涉及的所有表(行)都会加锁.
5.3. 事务嵌套
SQLAlchemy 中的事务嵌套有两种情况. 一是在 session 中管理的事务, 本身有层次性. 二是 session 和原始的 connection 之间, 是一种层次关系, 在这 session , connection 两个概念之中的事务同样具有这样的层次.
session 中的事务, 可能通过
begin_nested()方法做 savepoint :
session.add(u1) session.add(u2) session.begin_nested() session.add(u3) session.rollback() # rolls back u3, keeps u1 and u2 session.commit()
或者使用上下文对象:
for record in records: try: with session.begin_nested(): session.merge(record) except: print "Skipped record %s" % record session.commit()
嵌套的事务的一个效果, 是最外层事务提交整个变更才会生效.
user = User(name='2') session.begin_nested() session.add(user) session.commit() session.rollback()
于是, 前面说的第二种情况有一种应用方式, 就是在 connection 上做一个事务, 最终也在 connection 上回滚这个事务, 如果 session 是
bind到这个连接上的, 那么 session 上所做的更改全部不会生效:
conn = Engine.connect() session = Session(bind=conn) trans = conn.begin() user = User(name='2') session.begin_nested() session.add(user) session.commit() session.commit() trans.rollback()
在测试中这种方式可能会有用.
5.4. 二段式提交
二段式提交, Two-Phase, 是为解决分布式环境下多点事务控制的一套协议.
与一般事务控制的不同是, 一般事务是
begin, 之后
commit结束.
而二段式提交的流程上,
begin之后, 是
prepare transaction 'transaction_id', 这时相关事务数据已经持久化了. 之后, 再在任何时候(哪怕重启服务), 作
commit prepared 'transaction_id'或者
rollback prepared 'transaction_id'.
从多点事务的控制来看, 应用层要做的事是, 先把任务分发出去, 然后收集"事务准备"的状态(
prepare transaction的结果). 根据收集的结果决定最后是 commit 还是 rollback .
简单来说, 就是事务先保存, 再说提交的事.
SQLAlchemy 中对这个机制的支持, 是在构建会话类是加入
twophase参数:
Session = sessionmaker(twophase=True)
然后会话类可以根据一些策略, 绑定多个 Engine , 可以是多个数据库连接, 比如:
Session = sessionmaker(twophase=True)
Session.configure(binds={User: Engine, Blog: Engine2})
这样, 在获取一个会话实例之后, 就处在二段式提交机制的支持之下, SQLAlchemy 自己会作多点的协调了. 完整的流程:
Engine = create_engine('postgresql://test@localhost:5432/test', echo=True)
Engine2 = create_engine('postgresql://test@localhost:5432/test2', echo=True)
Session = sessionmaker(twophase=True)
Session.configure(binds={User: Engine, Blog: Engine2})
session = Session()
user = User(name='名字')
session.add(user)
session.commit()
对应的 SQL 大概就是:
begin; insert into "user" (name) values (?); prepare transaction 'xx'; commit prepared 'xx';
使用时, Postgresql 数据库需要把 max_prepared_transactions 这个配置项的值改成大于 0
6. SQLAlchemy字段类型
6.1. 基本类型
字段类型是在定义模型时, 对每个
Column的类型约定. 不同类型的字段类型在输入输出上, 及支持的操作方面, 有所区别.
这里只介绍
sqlalchemy.types.*中的类型, SQL 标准类型方面, 是写什么最后生成的 DDL 语句就是什么, 比如
BIGINT,
BLOG这些, 但是这些类型并不一定在所有数据库中都有支持. 除此而外, SQLAlchemy 也支持一些特定数据库的特定类型, 这些需要从具体的
dialects实现里导入.
- Integer/BigInteger/SmallInteger
- 整形.
- Boolean
- 布尔类型. Python 中表现为
True/False
, 数据库根据支持情况, 表现为BOOLEAN
或SMALLINT
. 实例化时可以指定是否创建约束(默认创建). - Date/DateTime/Time
(timezone=False)
- 日期类型,
Time
和DateTime
实例化时可以指定是否带时区信息. - Interval
- 时间偏差类型. 在 Python 中表现为
datetime.timedelta()
, 数据库不支持此类型则存为日期. - Enum
(*enums, **kw)
- 枚举类型, 根据数据库支持情况, SQLAlchemy 会使用原生支持或者使用
VARCHAR
类型附加约束的方式实现. 原生支持中涉及新类型创建, 细节在实例化时控制. - Float
- 浮点小数.
- Numeric
(precision=None, scale=None, decimal_return_scale=None, ...)
- 定点小数, Python 中表现为
Decimal
. - LargeBinary
(length=None)
- 字节数据. 根据数据库实现, 在实例化时可能需要指定大小.
- PickleType
- Python 对象的序列化类型.
- String
(length=None, collation=None, ...)
- 字符串类型, Python 中表现为
Unicode
, 数据库表现为VARCHAR
, 通常都需要指定长度. - Unicode
- 类似与字符串类型, 在某些数据库实现下, 会明确表示支持非 ASCII 字符. 同时输入输出也强制是
Unicode
类型. - Text
- 长文本类型, Python 表现为
Unicode
, 数据库表现为TEXT
. - UnicodeText
- 参考 Unicode .
7. SQLAlchemy混合属性机制
7.1. 直接行为
混合属性, 官方文档中称之为 Hybrid Attributes . 这种机制表现为, 一个属性, 在类和层面, 和实例的层面, 其行为是不同的. 之所以需要关注这部分的差异, 原因源于 Python 上下文和 SQL 上下文的差异.
类 层面经常是作为 SQL 查询时的一部分, 它面向的是 SQL 上下文. 而实例是已经得到或者创建的结果, 它面向的是 Python 上下文.
定义模型的 Column() 就是一个典型的混合属性. 作为实例属性时, 是具体的对象值访问, 而作为类属性时, 则有构成 SQL 语句表达式的功能.
class Interval(BaseModel): __tablename__ = 'interval' id = Column(Integer, autoincrement=True, primary_key=True) start = Column(Integer) end = Column(Integer) session.add(Interval(start=0, end=100)) session.commit()
实例行为:
ins = session.query(Interval).first() print (ins.end - ins.start)
类行为:
ins = session.query(Interval).filter(Interval.end - Interval.start > 10).first()
这种机制其实一直在被使用, 但是可能大家都没有留意一个属性在类和实例上的区别.
如果属性需要被进一步封装, 那么就需要明确声明 Hybrid Attributes 了:
from sqlalchemy.ext.hybrid import hybrid_property, hybrid_method class Interval(BaseModel): __tablename__ = 'interval' id = Column(Integer, autoincrement=True, primary_key=True) start = Column(Integer) end = Column(Integer) @hybrid_property def length(self): return self.end - self.start @hybrid_method def bigger(self, i): return self.length > i session.add(Interval(start=0, end=100)) session.commit() ins = session.query(Interval).filter(Interval.length > 10).first() ins = session.query(Interval).filter(Interval.bigger(10)).first() print( ins.bigger(1))
setter 的定义同样使用对应的装饰器即可:
class Interval(BaseModel): __tablename__ = 'interval' id = Column(Integer, autoincrement=True, primary_key=True) start = Column(Integer) end = Column(Integer) @hybrid_property def length(self): return abs(self.end - self.start) @length.setter def length(self, l): self.end = self.start + l
7.2. 表达式行为
前面说的属性, 在类和实例上有不同行为, 可以看到, 在类上的行为, 其实就是生成 SQL 表达式时的行为. 上面的例子只是简单的运算, SQLAlchemy 可以自动处理好 Python 函数和 SQL 函数的区别. 但是如果是一些特性更强的 SQL 函数, 就需要手动指定了. 于时, 这时的情况变成, 实例行为是 Python 范畴的调用行为, 而类行为则是生成 SQL 函数的相关表达式.
同时是前面的例子, 对于
length的定义, 更严格上来说, 应该是取绝对值的.
class Interval(BaseModel): __tablename__ = 'interval' id = Column(Integer, autoincrement=True, primary_key=True) start = Column(Integer) end = Column(Integer) @hybrid_property def length(self): return abs(self.end - self.start)
但是, 如果使用了 Python 的
abs()函数, 在生成 SQL 表达式时显示有无法处理了. 所以, 需要手动定义:
from sqlalchemy import func class Interval(BaseModel): __tablename__ = 'interval' id = Column(Integer, autoincrement=True, primary_key=True) start = Column(Integer) end = Column(Integer) @hybrid_property def length(self): return abs(self.end - self.start) @length.expression def length(self): return func.abs(self.end - self.start)
这样查询时就可以直接使用:
ins = session.query(Interval).filter(Interval.length > 1).first()
7.3. 应用于关系
总体上没有特别之处:
class Account(BaseModel):
__tablename__ = 'account'
id = Column(Integer, autoincrement=True, primary_key=True)
user = Column(Integer, ForeignKey('user.id'), index=True)
balance = Column(Integer, server_default='0')
class User(BaseModel):
__tablename__ = 'user'
id = Column(Integer, autoincrement=True, primary_key=True)
name = Column(Unicode(32), nullable=False, server_default='')
accounts = relationship('Account')
#balance = association_proxy('accounts', 'balance')
@hybrid_property
def balance(self):
return sum(x.balance for x in self.accounts)
查询时:
user = session.query(User).first() print (user.balance)
这里涉及的东西都是 Python 自己的, 包括那个
sum()函数, 和 SQL 没有关系.
如果想实现的是, 使用 SQL 的
sum()函数, 取出指定用户的总账户金额数, 那么就要考虑把
balance作成表达式的形式:
from sqlalchemy import select
@hybrid_property
def balance(self):
return select([func.sum(Account.balance)]).where(Account.user == self.id).label('balance_v')
#return func.sum(Account.balance)
这样的话,
User.balance只是单纯的一个表达式了, 查询时指定字段:
user = session.query(User, User.balance).first() print (user.balance_v)
注意, 如果写成:
session.query(User.balance).first()
意义就不再是"获取第一个用户的总金额", 而变成"获取总金额的第一个". 这里很坑吧.
像上面这样改, 实例层面就无法使用
balance属性. 所以, 还是先前介绍的, 表达式可以单独处理:
@hybrid_property
def balance(self):
return sum(x.balance for x in self.accounts)
@balance.expression
def balance(self):
return select([func.sum(Account.balance)]).where(Account.user == self.id).label('balance_v')
定义了表达式的
balance, 这部分作为查询条件上当然也是可以的:
user = session.query(User).filter(User.balance > 1).first()
内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理 
相关文章推荐
- 存储过程 与 事务管理
- Oracle截取字符串去掉字段末尾指定长度的字符
- 关于weka使用 ————连接数据库
- 关于Golang中database/sql包的学习笔记（转载）
- postgresql建库流程及其部分问题记录
- mysql 常用命令
- Redis的分片、预分片技术总结
- 分享oracle 获取12月的SQL
- Redis小结
- 编译安装Redis
- 详解Redis数据库配置不当遭恶意入侵过程
- memcache
- redis：Could not get a resource from the pool
- MySQL数据表基本操作
- 通过Proxool辅助数据库优化
- mysql主从复制小记
- memcached几个客户端测试
- MongoDB入门简介
- oracle提高查询效率24法
- redis keys命令之领悟