Jusene's Blog

Sqlalchemy ORM框架(一)

python
字数统计: 1.7k阅读时长: 9 min
2018/09/28 32 Share

sqlalchemy

sqlalchemy是python最常用的orm框架，ORM(对象关系映射)模式是一种为了解决面向对象与关系数据库存在的互不匹配的现象的技术,ORM是通过使用描述对象和数据库之间映射的元数据，将程序中的对象自动持久化到关系数据库中，ORM作为一层中间层，可以不必关系后端是什么数据库，整个转换过程由ORM完成。

使用

安装

1
~]# pip install sqlalchemy
1
from sqlalchemy import create_engine
2
3
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/mytest", echo=True)
4
engine.connect()  # 测试数据库可联通
5
engine.execute('select 1')    # 直接执行sql
6
<sqlalchemy.engine.result.ResultProxy at 0x109b53ac8>
7
list(engine.execute('select 1')
8
[(1,)]

定义映射关系

1
from sqlalchemy.ext.declarative import declarative_base
2
from sqlalchemy import Column, Integer, String
3
4
Model = declarative_base()
5
6
class User(Model):
7
    __tablename__ = "user"
8
9
    id = Column(Integer, primary_key=True, autoincrement=True)
10
    name = Column(String(64), nullable=False, unique=True)
11
    age = Column(Integer, nullable=False)
12
    
13
    def __repr__(self):
14
        return 'User(id={}, name={}, age={})'.format(self.id, self.name, self.age)
15
16
Model.metadata.create_all(engine) # 创建表
17
Model.metadata.drop_all(engine) # 删除表
18
19
User.__table__
20
Table('user', MetaData(bind=None), Column('id', Integer(), table=<user>, primary_key=True, nullable=False), Column('name', String(length=64), table=<user>, nullable=False), Column('age', Integer(), table=<user>, nullable=False), schema=None)

创建

1
student1 = User(name='jack', age=25)
2
student1.name
3
'jack'
4
5
## 创建一个session对话将数据写入数据库
6
from sqlalchemy.orm import sessionmaker
7
8
Session = sessionmaker(bind=engine)
9
session = Session()
10
session.add(student1)
11
session.commit()
12
13
# SQL: UPDATE user SET name='jusene' WHERE user.id = 1
14
student1.name = 'jusene'
15
session.add(student1)
16
session.commit()
17
18
19
## 一次写入多条数据
20
session.add_all([
21
    User(name='tom', age=12),
22
    User(name='jerry', age=23),
23
    User(name='pick', age=43),
24
])
25
session.new
26
IdentitySet([User(id=None, name=tom, age=12), User(id=None, name=jerry, age=23), User(id=None, name=pick, age=43)])
27
28
session.commit()

查询

1
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user ORDER BY user.id
2
for instance in session.query(User).order_by(User.id):
3
    print(instance.name, instance.age)
4
jusene 25
5
tom 12
6
jerry 23
7
pick 43
8
9
# SQL: SELECT user.name AS user_name, user.age AS user_age FROM user
10
for name, age in session.query(User.name, User.age):
11
    print(name, age)
12
jusene 25
13
tom 12
14
jerry 23
15
pick 43
16
17
# SQL: SELECT user.name AS name_label FROM user
18
for row in session.query(User.name.label('name_label')):
19
    print(row.name_label)
20
jerry
21
jusene
22
pick
23
tom
24
25
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user ORDER BY user.id LIMIT 1, 2
26
for u in session.query(User).order_by(User.id)[1:3]:
27
    print(u)
28
User(id=2, name=tom, age=12)
29
User(id=3, name=jerry, age=23)
30
31
for u in session.query(User).order_by(User.id).offset(1).limit(2):
32
    print(u)
33
User(id=2, name=tom, age=12)
34
User(id=3, name=jerry, age=23)
35
36
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user ORDER BY user.id DESC  LIMIT 1, 2
37
for u in session.query(User).order_by(User.id.desc()).offset(1).limit(2):
38
    print(u)
39
User(id=3, name=jerry, age=23)
40
User(id=2, name=tom, age=12)
41
42
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name = 'jusene'
43
for row in session.query(User).filter_by(name = 'jusene'):
44
    print(row)
45
User(id=1, name=jusene, age=25)
46
47
for row in session.query(User).filter(User.name == 'jusene'):
48
    print(row)
49
User(id=1, name=jusene, age=25)
50
51
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name != 'jusene'
52
for row in session.query(User).filter(User.name != 'jusene'):
53
    print(row)
54
User(id=2, name=tom, age=12)
55
User(id=3, name=jerry, age=23)
56
User(id=4, name=pick, age=43)
57
58
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name LIKE 'j%'
59
for row in session.query(User).filter(User.name.like('j%')):
60
    print(row)
61
User(id=3, name=jerry, age=23)
62
User(id=1, name=jusene, age=25)
63
64
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE lower(user.name) LIKE lower('j%')
65
for row in session.query(User).filter(User.name.ilike('j%')):
66
    print(row)
67
User(id=3, name=jerry, age=23)
68
User(id=1, name=jusene, age=25)
69
70
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name IN ('tom', 'jerry')
71
for row in session.query(User).filter(User.name.in_(['tom','jerry'])):
72
    print(row)
73
User(id=2, name=tom, age=12)
74
User(id=3, name=jerry, age=23)
75
76
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name IN (SELECT user.name AS user_name FROM user WHERE user.age IN (25, 12)
77
for row in session.query(User).filter(User.name.in_(session.query(User.name).filter(User.age.in_([25,12])))):
78
    print(row)
79
User(id=1, name=jusene, age=25)
80
User(id=2, name=tom, age=12)
81
82
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name NOT IN ('tom', 'jerry')
83
for row in session.query(User).filter(~User.name.in_(['tom','jerry'])):
84
    print(row)
85
User(id=1, name=jusene, age=25)
86
User(id=4, name=pick, age=43)
87
88
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name IS NULL
89
for row in session.query(User).filter(User.name == None):
90
    print(row)
91
92
for row in session.query(User).filter(User.name.is_(None)):
93
    print(row)
94
95
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name IS NOT NULL
96
for row in session.query(User).filter(User.name != None):
97
    print(row)
98
99
for row in session.query(User).filter(User.name.isnot_(None)):
100
    print(row)
101
102
# 与或非
103
from sqlalchemy import and_, or_, not_
104
105
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name = 'jusene' AND user.age = 25
106
for row in session.query(User).filter(and_(User.name == 'jusene', User.age == 25)):
107
    print(row)
108
109
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name = 'jusene' OR user.age = 25
110
for row in session.query(User).filter(or_(User.name == 'jusene', User.age == 12)):
111
    print(row)
112
113
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name != 'jusene' AND user.age = 12
114
for row in session.query(User).filter(and_(not_(User.name == 'jusene'), User.age == 12)):
115
    print(row)
116
117
# SQL: SELECT user.id AS user_id, user.name AS user_name, user.age AS user_age FROM user WHERE user.name BETWEEN 'tom' AND 'jerry'
118
for row in session.query(User).filter(User.name.between('tom','jerry')):
119
    print(row)
120
121
# SQL: SELECT user.age AS user_age, count(user.age) AS count_1 FROM user GROUP BY user.age
122
from sqlalchemy import func
123
for row in session.query(User.age, func.count(User.age)).group_by(User.age):
124
    print(row)
125
(12, 1)
126
(23, 1)
127
(25, 1)
128
(43, 1)

返回数据

  • all() 返回全部列表

    1
    session.query(User).all()
    2
    [User(id=1, name=jusene, age=25),
    3
     User(id=2, name=tom, age=12),
    4
     User(id=3, name=jerry, age=23),
    5
     User(id=4, name=pick, age=43)]

  • first() 返回第一条数据

    1
    session.query(User).first()
    2
    User(id=1, name=jusene, age=25)

  • one() 返回全部的数据，如果数据不止一条就会抛出异常

    1
    session.query(User).one()
    2
    MultipleResultsFound: Multiple rows were found for one()
    3
    4
    session.query(User).filter(User.name == 'jusene').one()
    5
    User(id=1, name=jusene, age=25)

  • one_or_none() 与one()类似，在无结果的情况下返回none

  • scalar() 调用one()方法，成功后返回行的第一列

删除

1
# DELETE FROM user WHERE user.name = 'jusene'
2
session.query(User).filter(User.name == 'jusene').delete()
3
session.rollback() # 可以撤销删除
4
session.commit() # 对数据库的修改必须commit()后生效，在这之前可以rollback撤销事务操作

更新

1
# UPDATE user SET name=%(name)s WHERE user.id = 1
2
student1 = session.query(User).filter(User.id == 1).one()
3
student1.name = 'jack'
4
session.add(student1)
5
session.commit()

使用文本SQL

1
from sqlalchemy import text
2
3
session.query(User).filter(text("id<:value and name=:name")).params(value=3, name='jusene').order_by(text('id')).one() 
4
User(id=1, name=jusene, age=25)
5
6
session.query(User).from_statement(text('SELECT * FROM user where name=:name')).params(name='jusene').one()
7
User(id=1, name=jusene, age=25)
8
9
stmt = text('SELECT name, id FROM user where name=:name')
10
stmt = stmt.columns(User.name, User.id)
11
session.query(User).from_statement(stmt).params(name='jusene').one()
12
User(id=1, name=jusene, age=25)

原文作者：Zhang Jusene

原文链接：http://jusene.github.io/2018/09/28/sqlalchemy/

发表日期：September 28th 2018, 12:40:50 am

更新日期：November 30th 2019, 1:54:24 am

版权声明：本文采用知识共享署名-非商业性使用 4.0 国际许可协议进行许可

CATALOG
  1. 1. sqlalchemy
  2. 2. 使用
  3. 3. 使用文本SQL
Total : 190
2020
2019
2018
2017
python boot troubshooting docker drbd ssh web ELK kubernetes Kubernetes git ansible tools dns corosync consul Django etcd security ftp Go bash bigdata grub heartbeat Java kickstart kvm lvs mongodb mysql mesh nc nfs mogilefs openresty upgrade puppet rabbitmq saltstack rsync screen socat svn tcpdump fastdfs haproxy iscsi keepalived cache openssl vpn optimization prometheus redis RPM cncf supervisord kernel JAVA
python Linux troubshooting Docker docker storage nginx Nosql kubernetes Flask Kubernetes ansible Shell cluster Django Go bigdata boot Java hypervisor mysql mesh tools openresty puppet rabbitmq saltstack optimization cncf varnish tomcat JAVA