Fork me on GitHub

sqlaLchemy--ORM常见操作

不做就别做,要做就坚持下去

记录下sqlalchemy中常用的操作,其它很多框架的ORM也是对sqlalchemy的包装,因此这些方法在框架中也是可用的

后面代码中使用的模型

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
from sqlalchemy import create_engine, Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, sessionmaker
from sqlalchemy.ext.declarative import declarative_base

engine = create_engine('mysql://root:123456@127.0.0.1:3306/test1?charset=utf8')
Base = declarative_base()


class User(Base):
__tablename__ = 'users'

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


class Address(Base):
__tablename__ = 'addresses'
id = Column(Integer, primary_key=True)
email_address = Column(String(100), nullable=False)
user_id = Column(Integer, ForeignKey('users.id'))

def __repr__(self):
return "<Address(email_address='%s')>" % self.email_address


Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

session = Session()

添加/更新/删除对象

添加对象

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> wu_user = User(name='wu',fullname='wu xxx xxxx',password='abcdef')
>>> session.add(wu_user)
>>> user = session.query(User).filter_by(name='wu').first()
>>> user
<User(name='wu', fullname='wu xxx xxxx', password='abcdef')>
>>> wu_user is user
True
>>> user.name
'wu'
>>> user.fullname
'wu xxx xxxx'
>>> user.password
'abcdef'
>>> session.commit()
>>>

更新对象

1
2
3
4
5
6
7
>>> wu_user.name = 'wu wu'
>>> session.add(wu_user)
>>> session.commit()
>>> user = session.query(User).filter_by(name='wu wu').first()
>>> user
<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>
>>>

删除

1
2
session.delete(jack)
session.commit()

查询

  1. 基本查询

    1
    2
    3
    4
    5
    6
    7
    8
    9
    >>> for instance in session.query(User).order_by(User.id):
    print(instance.name, instance.fullname)
    wu wu wu xxx xxxx
    hu hu xxx xxxx
    qi qi xxx xxxx
    hg hg xxx xxxx
    >>> print(str(session.query(User).order_by(User.id))) # 可以通过str查看执行的sql语句
    SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
    FROM users ORDER BY users.id
  2. 可以直接指定要查询的字段,此时查询返回的是一个KeyedTuple对象,可以当做元组那样对待:

    1
    2
    3
    4
    5
    6
    >>> for name, fullname in session.query(User.name, User.fullname):
    print(name, fullname)
    wu wu wu xxx xxxx
    hu hu xxx xxxx
    qi qi xxx xxxx
    hg hg xxx xxxx
  3. 通过lable为字段命别名:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    >>> for row in session.query(User.name.label('name_label')).all():
    print(row.name_label)
    wu wu
    hu
    qi
    hg
    >>> print(str(session.query(User.name.label('name_label'))))
    SELECT users.name AS name_label
    FROM users
    >>>
  4. aliased为实体命别名:

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    >>> from sqlalchemy.orm import aliased
    >>> user_alias = aliased(User, name='user_alias')
    >>> for row in session.query(user_alias, user_alias.name).all():
    print(row.user_alias)
    <User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>
    <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>
    <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>
    <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
    >>> print(str(session.query(user_alias, user_alias.name)))
    SELECT user_alias.id AS user_alias_id, user_alias.name AS user_alias_name, user_alias.fullname AS user_alias_fullname, user_alias.password AS user_alias_password
    FROM users AS user_alias
  5. limit offset 指定偏移和结果数目

    1
    2
    3
    4
    5
    6
    >>> session.query(User).order_by(User.id).limit(2).offset(1).all()
    [<User(name='hu', fullname='hu xxx xxxx', password='abcdef')>, <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>]
    >>> print(str(session.query(User).order_by(User.id).limit(2).offset(1)))
    SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
    FROM users ORDER BY users.id
    LIMIT %s, %s
  6. Query对象是完全生成的,他返回的是一个新的Query对象,可继续使用查询语句

1
2
3
4
5
6
7
8
9
10
11
>>> for user in session.query(User).\
filter(User.name=='hu').\
filter(User.fullname=='hu xxx xxxx'):
print(user)


<User(name='hu', fullname='hu xxx xxxx', password='abcdef')>
>>> print(str(session.query(User).filter(User.name=='hu').filter(User.fullname=='hu xxx xxxx')))
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 = %s AND users.fullname = %s

过滤

  1. equals/not equals
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# euqals
>>> session.query(User).filter(User.name == 'wu wu').first()
<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>
>>> print(str(session.query(User).filter(User.name == 'wu wu')))
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 = %s

# not equals
>>> session.query(User).filter(User.name != 'wu wu').all()
[<User(name='hu', fullname='hu xxx xxxx', password='abcdef')>, <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>, <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(User.name != 'wu wu')))
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 != %s
  1. like/ilike
1
2
3
4
5
6
7
# like(在有的后端不区分大小写,有的区分大小写,如果要不区分大小写,最好用ilike)
>>> session.query(User).filter(User.name.like('%wu%')).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(User.name.like('%wu%'))))
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 LIKE %s
  1. in/not in
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
# in
>>> session.query(User).filter(User.name.in_(['wu wu','hu'])).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>, <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(User.name.in_(['wu wu','hu']))))
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 IN (%s, %s)

# not in
>>> session.query(User).filter(~User.name.in_(['wu wu','hu'])).all()
[<User(name='qi', fullname='qi xxx xxxx', password='abcdef')>, <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(~User.name.in_(['wu wu','hu']))))
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 NOT IN (%s, %s)
>>>
  1. is null/is not null
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# is null
>>> session.query(User).filter(User.name == None).all()
[<User(name='None', fullname='None', password='None')>]
>>> print(str(session.query(User).filter(User.name == None)))
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 IS NULL
>>> session.query(User).filter(User.name.is_(None)).all()
[<User(name='None', fullname='None', password='None')>]

# is not null
>>> session.query(User).filter(User.name != None).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>, <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>, <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>, <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>]
>>> session.query(User).filter(User.name.isnot(None)).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>, <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>, <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>, <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(User.name.isnot(None))))
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 IS NOT NULL
  1. and/or
1
2
3
4
5
6
7
8
9
10
11
12
# and
>>> from sqlalchemy import and_
>>> session.query(User).filter(User.name == 'wu wu',User.fullname == 'wu xxx xxxx').all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]
>>> session.query(User).filter(and_(User.name == 'wu wu',User.fullname == 'wu xxx xxxx')).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]
>>> session.query(User).filter(User.name == 'wu wu').filter(User.fullname == 'wu xxx xxxx').all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]

# or
>>> session.query(User).filter(or_(User.name == 'wu wu', User.name == 'hu')).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>, <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>]
  1. match/contains
1
2
3
4
5
6
7
# 有的后端数据库可能不支持
>>> session.query(User).filter(User.name.contains('u w')).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).filter(User.name.contains('u w'))))
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 LIKE concat(concat('%%', %s), '%%'))

使用文本SQL

Query通过指定它们与text()结合,可以灵活地使用文字字符串 ,这可以被大多数适用的方法所接受。例如, filter()order_by()

1
2
3
4
5
6
7
8
>>> for user in session.query(User).filter(text("id<5")).order_by(text("id")).all():
print(user.name)


wu wu
hu
qi
hg

绑定参数可以使用冒号在基于字符串的SQL中指定。要指定值,可以使用params()方法:

1
2
3
4
5
6
7
8
>>> session.query(User).filter(text("id<:value and name=:name")). \
params(value=2, name='wu wu').order_by(User.id).first()
<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>
>>> print(str(session.query(User).filter(text("id<:value and name=:name")). \
params(value=2, name='wu wu').order_by(User.id)))
SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
WHERE id<%s and name=%s ORDER BY users.id

名称上的列匹配适用于简单情况,但在处理包含重复列名称的复杂语句时或在使用不易与特定名称匹配的匿名化ORM结构时会变得很难处理。另外,在处理结果行时,我们可能会发现映射列中存在打字行为。对于这些情况,该text()构造允许我们将其文本SQL位置链接到Core或ORM映射列表达式; 可以通过将列表达式作为位置参数传递给TextClause.columns()方法来实现此目的:

1
2
3
4
5
6
7
>>> stmt = text("SELECT name, id, fullname, password FROM users where name=:name")
>>> stmt = stmt.columns(User.name, User.id, User.fullname, User.password)
>>> session.query(User).from_statement(stmt).params(name='wu wu').all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>]
>>> print(str(session.query(User).from_statement(stmt).params(name='wu wu')))
SELECT name, id, fullname, password FROM users where name=%s
>>>

计数

Query有一个便利的计数方法count()

1
2
>>> session.query(User).filter(User.name.like('%wu')).count()
1

对于需要特别指出“要计数的东西”的情况,我们可以直接使用表达式指定“计数”函数func.count(),可从func构造中获得:

1
2
3
4
5
6
>>> from sqlalchemy import func
>>> session.query(func.count(User.name), User.name).group_by(User.name).all()
[(0, None), (1, 'hg'), (1, 'hu'), (1, 'qi'), (1, 'wu wu')]
>>> print(str(session.query(func.count(User.name), User.name).group_by(User.name)))
SELECT count(users.name) AS count_1, users.name AS users_name
FROM users GROUP BY users.name

SELECT count(*) FROM table写法:

1
2
3
4
5
>>> session.query(func.count('*')).select_from(User).scalar()
5
>>> print(str(session.query(func.count('*')).select_from(User)))
SELECT count(%s) AS count_1
FROM users

相关对象

1
2
3
4
5
6
7
8
9
10
11
12
13
>>> user
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
>>> address1 = Address(email_address='1111111111@xx.com',user=user)
>>> address2 = Address(email_address='2222222222@xx.com',user=user)
>>> session.add_all([address1,address2])
>>> session.commit()
>>> user.addresses
[<Address(email_address='1111111111@xx.com')>, <Address(email_address='2222222222@xx.com')>]
>>> address1.user
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
>>> address2.user
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
>>>

在User上执行查询时并不会将addresses也查询出来

1
2
3
4
5
6
>>> session.query(User).all()
[<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')>, <User(name='hu', fullname='hu xxx xxxx', password='abcdef')>, <User(name='qi', fullname='qi xxx xxxx', password='abcdef')>, <User(name='hg', fullname='hg xxx xxxx', password='abcdef')>, <User(name='None', fullname='None', password='None')>]
>>> print(session.query(User))
SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
FROM users
>>>

只有当查询user.addresses集合时才会加载,因为使用的是惰性加载

连接查询

User和之间的简单隐式连接Address,我们可以使用Query.filter()它们将相关的列等同起来:

1
2
3
4
5
6
7
8
9
>>> for u, a in session.query(User, Address).\
filter(User.id==Address.user_id).\
all():
print(u,a)


<User(name='hg', fullname='hg xxx xxxx', password='abcdef')> <Address(email_address='1111111111@xx.com')>
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')> <Address(email_address='2222222222@xx.com')>
>>>

使用Query.join()也可以实现上面的查询:

1
2
3
4
5
6
7
>>> for u, a in session.query(User,Address).join(Address).all():
print(u,a)


<User(name='hg', fullname='hg xxx xxxx', password='abcdef')> <Address(email_address='1111111111@xx.com')>
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')> <Address(email_address='2222222222@xx.com')>
>>>

Query.join()知道如何连接UserAddress因为它们之间只有一个外键。如果没有外键或多个外键,Query.join()则在使用下列其中一种形式时效果更好:

1
2
3
4
query.join(Address, User.id==Address.user_id)    # explicit condition
query.join(User.addresses) # specify relationship from left to right
query.join(Address, User.addresses) # same, with explicit target
query.join('addresses') # same, using a string

Query如果有多个实体,选择什么?
Query.join()方法通常会从实体列表中最左边的项加入,省略ON子句或ON子句是普通SQL表达式时。要控制JOIN列表中的第一个实体,使用Query.select_from()方法:

1
query = session.query(User, Address).select_from(Address).join(User)

子查询

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
>>> from sqlalchemy.sql import func
>>> stmt = session.query(Address.user_id, func.count('*').\
label('address_count')).\
group_by(Address.user_id).subquery()
>>>
>>> for u, count in session.query(User, stmt.c.address_count).\
outerjoin(stmt, User.id==stmt.c.user_id).order_by(User.id):
print(u, count)


<User(name='wu wu', fullname='wu xxx xxxx', password='abcdef')> None
<User(name='hu', fullname='hu xxx xxxx', password='abcdef')> None
<User(name='qi', fullname='qi xxx xxxx', password='abcdef')> None
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')> 2
<User(name='None', fullname='None', password='None')> None
>>> print(str(session.query(User, stmt.c.address_count).\
outerjoin(stmt, User.id==stmt.c.user_id).order_by(User.id)))
SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, anon_1.address_count AS anon_1_address_count
FROM users LEFT OUTER JOIN (SELECT addresses.user_id AS user_id, count(%s) AS address_count
FROM addresses GROUP BY addresses.user_id) AS anon_1 ON users.id = anon_1.user_id ORDER BY users.id
>>>

如果希望子查询映射到实体?可以使用aliased() 将映射类的“别名”关联到子查询:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> stmt = session.query(Address).\
filter(Address.email_address != 'j25@yahoo.com').\
subquery()
>>>
>>> adalias = aliased(Address, stmt)
>>> for user, address in session.query(User, adalias).\
join(adalias, User.addresses):
print(user)
print(address)


<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
<Address(email_address='1111111111@xx.com')>
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
<Address(email_address='2222222222@xx.com')>

常见关系运算符

  • __eq__()(多对一“等于”比较):

    query.filter(Address.user == someuser)

  • __ne__()(多对一“不等于”比较):

    query.filter(Address.user != someuser)

  • IS NULL(多对一比较,也可使用__eq__())

  • query.filter(Address.user == None)

  • contains()(用于一对多集合):

    query.filter(User.addresses.contains(someaddress))

  • any()(用于集合):

    1
    2
    3
    4
    query.filter(User.addresses.any(Address.email_address == 'bar'))

    # also takes keyword arguments:
    query.filter(User.addresses.any(email_address='bar'))
  • has()(用于标量参考):

    query.filter(Address.user.has(name='ed'))

  • Query.with_parent()(用于任何关系):

    session.query(Address).with_parent(someuser, 'addresses')

急切加载

如果没有指定关系加载的方式,默认是延迟加载,例如上面的User.addresses,如果要在查询时加载关系,需要使用Query.options()指定加载的方式

子查询加载

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
>>> from sqlalchemy.orm import subqueryload
>>> hg = session.query(User).\
options(subqueryload(User.addresses)).\
filter_by(name='hg').one()

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 = ?
SELECT addresses.id AS addresses_id,
addresses.email_address AS addresses_email_address,
addresses.user_id AS addresses_user_id,
anon_1.users_id AS anon_1_users_id
FROM (SELECT users.id AS users_id
FROM users WHERE users.name = ?) AS anon_1
JOIN addresses ON anon_1.users_id = addresses.user_id
ORDER BY anon_1.users_id, addresses.id
>>> hg
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
>>> hg.addresses
[<Address(email_address='1111111111@xx.com')>, <Address(email_address='2222222222@xx.com')>]

加入载入

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
>>> from sqlalchemy.orm import joinedload
>>> hg = session.query(User). \
options(joinedload(User.addresses)).\
filter(User.name == 'hg').one()
>>>
>>> hg
<User(name='hg', fullname='hg xxx xxxx', password='abcdef')>
>>> hg.addresses
[<Address(email_address='1111111111@xx.com')>, <Address(email_address='2222222222@xx.com')>]
>>> print(str(session.query(User). \
options(joinedload(User.addresses)).\
filter(User.name == 'hg')))
SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password, addresses_1.id AS addresses_1_id, addresses_1.email_address AS addresses_1_email_address, addresses_1.user_id AS addresses_1_user_id
FROM users LEFT OUTER JOIN addresses AS addresses_1 ON users.id = addresses_1.user_id
WHERE users.name = %s

参考

http://docs.sqlalchemy.org/en/latest/orm/

-------------本文结束感谢您的阅读-------------

本文标题:sqlaLchemy--ORM常见操作

文章作者:Longofo

发布时间:2018年06月28日 - 23:06

最后更新:2018年07月09日 - 19:07

原始链接:http://longofo.cc/sqlaLchemy--ORM常见操作.html

许可协议: 署名-非商业性使用-禁止演绎 4.0 国际 转载请保留原文链接及作者。

请我吃包辣条也好啊!!!
分享到: