SQLAlchemy 学习笔记（一）：Engine 与 SQL 表达式语言

from sqlalchemy import create_engine

engine = create_engine('sqlite:///:memory:',
                    echo=True,  # echo=True 表示打印出自动生成的 SQL 语句（通过 logging）
                    pool_size=5,  # 连接池容量，默认为 5，生产环境下太小，需要修改。
                    # 下面是 connection 回收的时间限制，默认 -1 不回收
                    pool_recycle=7200)   # 超过 2 小时就重新连接（MySQL 默认的连接最大闲置时间为 8 小时）

# PostgreSQL
postgresql+psycopg2://scott:tiger@localhost/dbtest

# MySQL + PyMySQL（或者用更快的  mysqlclient）
mysql+pymysql://scott:tiger@localhost/dbtest

# sqlite 内存数据库
# 注意 sqlite 要用三个斜杠，表示不存在 hostname，sqlite://<nohostname>/<path>
sqlite:///:memory:

# sqlite 文件数据库
# 四个斜杠是因为文件的绝对路径以 / 开头：/home/ryan/Codes/Python/dbtest.db
sqlite:////home/ryan/Codes/Python/dbtest.db

# SQL Server + pyodbc
# 首选基于 dsn 的连接，dsn 的配置请搜索hhh
mssql+pyodbc://scott:tiger@some_dsn

with engine.connect() as conn:
    res = conn.execute("select username from users")  # 无参直接使用

    # 使用问号作占位符，前提是下层的 DBAPI 支持。更好的方式是使用 text()，这个后面说
    conn.execute("INSERT INTO table (id, value) VALUES (?, ?)", 1, "v1")  # 参数不需要包装成元组

    # 查询返回的是 ResultProxy 对象，有和 dbapi 相同的 fetchone()、fetchall()、first() 等方法，还有一些拓展方法
    for row in result:
        print("username:", row['username'])

def transaction_a(connection):
    trans = connection.begin()  # 开启一个 transaction
    try:
        # do sthings
        trans.commit()  # 这里需要手动提交
    except:
        trans.rollback()  # 出现异常则 rollback
        raise
# do other things

with engine.connect() as conn:
    transaction_a(conn)

# 1. 参数绑定语法
from sqlalchemy import text

result = connection.execute(
            # 使用 :key 做占位符
            text('select * from table where id < :id and typeName=:type'),
            {'id': 2,'type':'USER_TABLE'})  # 用 dict 传参数，更易读

# 2. 参数类型指定
from sqlalchemy import DateTime

date_param=datetime.today()+timedelta(days=-1*10)
sql="delete from caw_job_alarm_log  where alarm_time < :alarm_time_param"

# bindparams 是 bindparam 的列表，bindparam 则提供参数的一些额外信息（类型、值、限制等）
t=text(sql, bindparams=[bindparam('alarm_time_param', type_=DateTime, required=True)])
connection.execute(t, {"alarm_time_param": date_param})

stmt = text("SELECT * FROM table",
            # 使用 typemap 指定将 id 列映射为 Integer 类型，name 映射为 String 类型
            typemap={'id': Integer, 'name': String},
          )
result = connection.execute(stmt)
# 对多个查询结果，可以用 for obj in result 遍历
# 也可用 fetchone() 只获取一个

from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey

metadata = MetaData()  # 先创建元数据（收藏夹）

users = Table('user', metadata,  # 创建 user 表，并放到 metadata 中
              Column('id', Integer, primary_key=True),
              Column('name', String),
              Column('fullname', String)
             )

addresses = Table('address', metadata,
                  Column('id', Integer, primary_key=True),
                  Column('user_id', None, ForeignKey('user.id')),  # 外键约束，引用 user 表的 id 列
                  Column('email_address', String, nullable=False)
                 )

metadata.create_all(engine)  # 使用 engine 创建 metadata 内的所有 Tables（会检测表是否已经存在，所以可以重复调用）

item = Table('item', metadata,  # 商品 table
    Column('id', Integer, primary_key=True),
    Column('name', String(60), nullable=False),
    Column('invoice_id', Integer, nullable=False),  # 发票 id，是外键
    Column('ref_num', Integer, nullable=False),
    ForeignKeyConstraint(['invoice_id', 'ref_num'],  # 当前表中的外键名称
                         ['invoice.id', 'invoice.ref_num'])  # 被引用的外键名称的序列（被引用的表）
)

from sqlalchemy import PrimaryKeyConstraint

my_table = Table('mytable', metadata,
            Column('id', Integer),
            Column('version_id', Integer),
            Column('data', String(50)),
            PrimaryKeyConstraint('id', 'version_id', name='mytable_pk')
        )

# 方法一，使用 values 传参
ins = users.insert().values(name="Jack", fullname="Jack Jones")  # 可以通过 str(ins) 查看自动生成的 sql
connection.execute(ins)

# 方法二，参数传递给 execute()
conn.execute(users.insert(), id=2, name='wendy', fullname='Wendy Williams')

# 方法三，批量 INSERT，相当于 executemany
conn.execute(addresses.insert(), [  # 插入到 addresses 表
    {'user_id': 1, 'email_address': '[email protected]'},  # 传入 dict 列表
    {'user_id': 1, 'email_address': '[email protected]'},
    {'user_id': 2, 'email_address': '[email protected]'},
    {'user_id': 2, 'email_address': '[email protected]'}
])

# 此外，通过使用 bindparam，INSERT 还可以执行更复杂的操作
stmt = users.insert() \
         .values(name=bindparam('_name') + " .. name")  # string 拼接
conn.execute(stmt, [
        {'id':4, '_name':'name1'},
        {'id':5, '_name':'name2'},
        {'id':6, '_name':'name3'},
     ])

_table.delete() \
        .where(_table.c.f1==value1) \
        .where(_table.c.f2==value2)  # where 指定条件

# 举例
stmt = users.update() \
             .where(users.c.name == 'jack') \
             .values(name='tom')
conn.execute(stmt)

# 批量更新
stmt = users.update() \
             .where(users.c.name == bindparam('oldname')) \
             .values(name=bindparam('newname'))
conn.execute(stmt, [
     {'oldname':'jack', 'newname':'ed'},
     {'oldname':'wendy', 'newname':'mary'},
     {'oldname':'jim', 'newname':'jake'},
     ])

from sqlalchemy.sql import select

# 1. 字段选择
s1 = select([users])  # 相当于 select * from users
s2 = select([users.c.name, users.c.fullname])  # 这个就是只 select 一部分

# 2. 添加过滤条件
s3 = select([users]) \
    .where(users.c.id == addresses.c.user_id)

res = conn.execute(s1)
# 可用 for row in res 遍历结果集，也可用 fetchone() 只获取一行

row1 = result.fetchone()  # 对应 cursor.fetchone()
row2 = result.fetchall()  # 对应 cursor.fetchall()
row3 = result.fetchmany(size=3)  # 对应 cursor.fetchmany(size=3)

row4 = result.first()  # 获取一行，然后立即调用 result 的 close() 方法

col = row[mytable.c.mycol] # 获取 mycol 这一列

result.rowcount  # 结果集的行数

同时，result 也实现了 next protocol，因此可以直接用 for 循环遍历

from sqlalchemy.sql import and_, or_, not_

s = select([(users.c.fullname +
               ", " + addresses.c.email_address).
                label('title')]).\
        where(users.c.id == addresses.c.user_id).\
        where(users.c.name.between('m', 'z')).\
        where(
               or_(
                  addresses.c.email_address.like('%@aol.com'),
                  addresses.c.email_address.like('%@msn.com')
               )
        )