sqlalchemy 的 Core 方式使用示例

知乎： sqlalchemy 的 Core 方式操作数据是一种怎样的体验？

答： 爽！

本文基于：win 10 + python 3.4 + sqlalchemy 1.0.13

基本步骤如下：

1. 绑定数据库

from sqlalchemy import create_engine

engine = create_engine(‘sqlite:///:memory:‘, echo=True)

2. 连接数据库

conn = engine.connect()

3. 元数据

from sqlalchemy import MetaData

metadata = MetaData(engine)

4. 定义表

from sqlalchemy import Table, Column, Integer, String, ForeignKey, Sequence
users = Table(‘users‘, metadata,
    Column(‘id‘, Integer, Sequence(‘user_id_seq‘), primary_key=True),
    Column(‘name‘, String),
    Column(‘fullname‘, String),
)

addresses = Table(‘addresses‘, metadata,
    Column(‘id‘, Integer, primary_key=True),
    Column(‘user_id‘, None, ForeignKey(‘users.id‘)),
    Column(‘email_address‘, String, nullable=False)
)

5. 创建表

# metadata.drop_all()metadata.create_all()

6. 插入

# 方式一
ins = users.insert().values(name=‘jack‘, fullname=‘Jack Jones‘)
conn.execute(ins)

# 方式二
conn.execute(users.insert(), id=2, name=‘wendy‘, fullname=‘Wendy Williams‘)

# 方式三
conn.execute(addresses.insert(), [
    {‘user_id‘: 1, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 1, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 2, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 2, ‘email_address‘ : ‘[email protected]‘},
 ])

7. 查询

from sqlalchemy.sql import select
for row in conn.execute(select([users])):
    print("name:", row[users.c.name], "; fullname:", row[users.c.fullname])

for row in conn.execute(select([users, addresses])):
    print(row)

for row in conn.execute(select([users, addresses]).where(users.c.id == addresses.c.user_id)):
    print(row)

from sqlalchemy.sql import and_, or_, not_
s = select([(users.c.fullname +
              ", " + addresses.c.email_address).
               label(‘title‘)]).       where(
          and_(
              users.c.id == addresses.c.user_id,
              users.c.name.between(‘m‘, ‘z‘),
              or_(
                 addresses.c.email_address.like(‘%@aol.com‘),
                 addresses.c.email_address.like(‘%@msn.com‘)
              )
          )
       )
conn.execute(s).fetchall()

8. 完整代码

# 绑定数据库
from sqlalchemy import create_engine
engine = create_engine(‘sqlite:///:memory:‘, echo=True)

# 连接数据库
conn = engine.connect()

# 元数据
from sqlalchemy import MetaData
metadata = MetaData(engine)

# 定义表
from sqlalchemy import Table, Column, Integer, String, ForeignKey, Sequence
users = Table(‘users‘, metadata,
    Column(‘id‘, Integer, Sequence(‘user_id_seq‘), primary_key=True),
    Column(‘name‘, String),
    Column(‘fullname‘, String),
)

addresses = Table(‘addresses‘, metadata,
    Column(‘id‘, Integer, primary_key=True),
    Column(‘user_id‘, None, ForeignKey(‘users.id‘)),
    Column(‘email_address‘, String, nullable=False)
)

# 创建表# metadata.drop_all()metadata.create_all()

# 插入
# 方式一
ins = users.insert().values(name=‘jack‘, fullname=‘Jack Jones‘)
result = conn.execute(ins)
# 方式二
conn.execute(users.insert(), id=2, name=‘wendy‘, fullname=‘Wendy Williams‘)
# 方式三
conn.execute(addresses.insert(), [
    {‘user_id‘: 1, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 1, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 2, ‘email_address‘ : ‘[email protected]‘},
    {‘user_id‘: 2, ‘email_address‘ : ‘[email protected]‘},
 ])

# 查询
from sqlalchemy.sql import select
for row in conn.execute(select([users])):
    print("name:", row[users.c.name], "; fullname:", row[users.c.fullname])

for row in conn.execute(select([users, addresses])):
    print(row)

for row in conn.execute(select([users, addresses]).where(users.c.id == addresses.c.user_id)):
    print(row)

from sqlalchemy.sql import and_, or_, not_
s = select([(users.c.fullname +
              ", " + addresses.c.email_address).
               label(‘title‘)]).       where(
          and_(
              users.c.id == addresses.c.user_id,
              users.c.name.between(‘m‘, ‘z‘),
              or_(
                 addresses.c.email_address.like(‘%@aol.com‘),
                 addresses.c.email_address.like(‘%@msn.com‘)
              )
          )
       )
conn.execute(s).fetchall()