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

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

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

metadata = MetaData('mysql+pymysql://root:[e-mail]/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

结果如图:

---

数据库引擎:

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:[e-mail]/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

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:[e-mail]/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

如图:

---