Python的ORM框架SQLAlchemy使用入门(二)【连接MySql数据库】

连接，增加数据 代码如下：

from sqlalchemy import *
from datetime import datetime
from sqlalchemy.orm import *

metadata = MetaData('mysql+pymysql://root:123456@localhost/SQLAlchemySample')
metadata.bind.echo = False

user_table = Table(
'tf_user', metadata,
Column('id', Integer, primary_key=True),
Column('user_name', Unicode(16), unique=True, nullable=False),
Column('password', Unicode(40), nullable=False),
Column('display_name', Unicode(255), default=''),
Column('created', DateTime, default=datetime.now())
)

group_table = Table(
'tf_group', metadata,
Column('id', Integer, primary_key=True),
Column('group_name', Unicode(16), unique=True, nullable=False),
)

permission_table = Table(
'tf_permission', metadata,
Column('id', Integer, primary_key=True),
Column('permission_name', Unicode(16), unique=True, nullable=False)
)

user_group_table = Table(
'tf_user_group', metadata,
Column('user_id', None, ForeignKey('tf_user.id'), primary_key=True),
Column('group_id', None, ForeignKey('tf_group.id'), primary_key=True)
)

group_permission_table = Table(
'tf_group_permission', metadata,
Column('permission_id', None, ForeignKey('tf_permission.id'), primary_key=True),
Column('group_id', None, ForeignKey('tf_group.id'), primary_key=True)
)

metadata.create_all()

class User(object): pass
class Group(object): pass
class Permission(object): pass

mapper(User, user_table)
mapper(Group, group_table)
mapper(Permission, permission_table)

if __name__ == '__main__':

"""
连接池
"""
user_table.delete().execute()

user_table.insert().execute(user_name='rick1', password='secret', display_name='rick C')
user_table.insert().execute(user_name='rick2', password='secret', display_name='rick C')
result = user_table.select().execute()
for row in result:
print(row)

result = user_table.select().execute()
row = result.fetchone()
print(row['user_name'])

"""
改变session
"""
print("##############################")
Session = sessionmaker()
session = Session()

query = session.query(User)
print("User列表:",list(query))
for user in query:
print("用户名称:"+user.user_name)

for user in query.filter(User.user_name.like("rick%")):
print('({0}, {1}, {2})'.format(user.id, user.user_name, user.created))

newuser = User()
newuser.user_name = 'yoshiya'
newuser.password = 'hoge'
session.add(newuser)
session.commit()
print("#################")

# auto flush
for user in query:
print(user.user_name)

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
[/code]

结果如图:

数据库引擎:

from sqlalchemy import *
from datetime import datetime
from sqlalchemy.orm import *
import logging

if __name__ == '__main__':
settings = {
'echo': True,
'echo_pool': True,
'encoding': 'utf-8',
'pool_size': 128,
'strategy': 'threadlocal'
}
url = "mysql+pymysql://root:123456@localhost/SQLAlchemySample"
engine = create_engine(url, **settings)
handler= logging.FileHandler('sqlalchemy.log')
handler.level = logging.DEBUG
logging.getLogger('sqlalchemy.engine').addHandler(handler)
logging.getLogger('sqlalchemy.pool').addHandler(handler)
#logging.getLogger('sqlalchemy.orm').addHandler(handler)

conn = engine.connect()
result = conn.execute('select user_name from tf_user')
for r in result:
print(r)
conn.close()

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
[/code]

DML

import sqlalchemy as sa
from sqlalchemy import MetaData, Table, Column, Integer, String, ForeignKey, intersect
from sqlalchemy import create_engine, bindparam

if __name__ == '__main__':
"""1.定义元信息，绑定到引擎"""
metadata = MetaData()
# 引擎绑定
engine = create_engine('mysql+pymysql://root:123456@localhost/SQLAlchemySample')
metadata.bind = engine

"""2.创建表格，初始化数据库"""
simple_table = Table('simple', metadata,
Column('id', Integer, primary_key=True),
Column('col1', String(20))
)
second_table = Table('second', metadata,
Column('id', Integer, primary_key=True),
Column('simple_id', Integer, ForeignKey('simple.id'), primary_key=True),
)

stmt = simple_table.insert()
print("stmt：",stmt)
print("params：",stmt.compile().params)

simple_table.delete(bind=engine).execute()

# 创建一个表
#simple_table.create(bind=engine)
#second_table.create(bind=engine)

engine.execute(stmt, col1='Foo')

stmt = simple_table.insert(values=dict(col1='new data'))
stmt.execute()
print(stmt.compile().params)

# multi insert , stmt -> metadata -> engine
stmt.execute([dict(col1='1'), dict(col1='2'), dict(col1='3')])

# 更新
stmt = simple_table.update( whereclause="id=2", values=dict(col1='update data'))
print(stmt)
#stmt.execute()
# 删除
stmt = simple_table.delete(whereclause="id='18'")
print(stmt)
#stmt.execute()

# select
#stmt = simple_table.select(whereclause='id="8"')
print('============================')
stmt = simple_table.select(simple_table.c.id=='7')
print(stmt)
print(stmt.execute().fetchone())
stmt = simple_table.select(simple_table.c.col1!='update data')
print(stmt.execute().fetchall())
print(stmt.execute().rowcount)

#绑定参数
print('============================')
stmt = simple_table.select(whereclause=simple_table.c.id==bindparam('id'))
print(stmt.execute(id=6).fetchone())
stmt = simple_table.select(group_by=[simple_table.c.col1])
print(stmt.execute().fetchall())

print('============================')
#连接
from_obj = simple_table.join(second_table)
#from_obj = simple_table.outerjoin(second_table)
q = simple_table.select().select_from(from_obj).where(simple_table.c.id == second_table.c.simple_id)
print(q.column('second.simple_id'))
print(q.execute().fetchall())

#设置操作
print('============================')
q1 = simple_table.select(simple_table.c.id > 1)
q2 = simple_table.select(simple_table.c.id < 7)
print(q1.execute().fetchall())
print(q2.execute().fetchall())
q = intersect(q1, q2)
print(q)

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
[/code]

如图: