SQLAlchemy 教程 —— 进阶篇
使用
首先是連接到數據庫，SQLALchemy支持多個數據庫引擎，不同的數據庫引擎連接字符串不一樣，常用的有
mysql://username:password@hostname/database postgresql://username:password@hostname/database sqlite:absolute/path/to/database sqlite:///c:/absolute/path/to/database更多連接字符串的介紹參見這里
下面是連接和使用sqlite數據庫的例子
1. connection
使用傳統的connection的方式連接和操作數據庫
from sqlalchemy import create_engine# 數據庫連接字符串 DB_CONNECT_STRING = 'sqlite:///:memory:'# 創建數據庫引擎,echo為True,會打印所有的sql語句 engine = create_engine(DB_CONNECT_STRING, echo=True)# 創建一個connection，這里的使用方式與python自帶的sqlite的使用方式類似 with engine.connect() as con:# 執行sql語句，如果是增刪改，則直接生效，不需要commitrs = con.execute('SELECT 5')data = rs.fetchone()[0]print "Data: %s" % dataSQLAlchemy是python的一個數據庫ORM工具，提供了強大的對象模型間的轉換，可以滿足絕大多數數據庫操作的需求，并且支持多種數據庫引擎（sqlite，mysql，postgres, mongodb等），在這里記錄基本用法和學習筆記
一、安裝
通過pip安裝
$ pip install SQLAlchemy二、使用
首先是連接到數據庫，SQLALchemy支持多個數據庫引擎，不同的數據庫引擎連接字符串不一樣，常用的有
mysql://username:password@hostname/database postgresql://username:password@hostname/database sqlite:absolute/path/to/database sqlite:///c:/absolute/path/to/database更多連接字符串的介紹參見這里
下面是連接和使用sqlite數據庫的例子
1. connection
使用傳統的connection的方式連接和操作數據庫
from sqlalchemy import create_engine# 數據庫連接字符串 DB_CONNECT_STRING = 'sqlite:///:memory:'# 創建數據庫引擎,echo為True,會打印所有的sql語句 engine = create_engine(DB_CONNECT_STRING, echo=True)# 創建一個connection，這里的使用方式與python自帶的sqlite的使用方式類似 with engine.connect() as con:# 執行sql語句，如果是增刪改，則直接生效，不需要commitrs = con.execute('SELECT 5')data = rs.fetchone()[0]print "Data: %s" % data與python自帶的sqlite不同，這里不需要Cursor光標，執行sql語句不需要commit
2. connection事務
使用事務可以進行批量提交和回滾
from sqlalchemy import create_engine# 數據庫連接字符串 DB_CONNECT_STRING = 'sqlite:Users/zhengxiankai/Desktop/Document/db.sqlite' engine = create_engine(DB_CONNECT_STRING, echo=True)with engine.connect() as connection:trans = connection.begin()try:r1 = connection.execute("select * from User")r2 = connection.execute("insert into User(name, age) values(?, ?)", 'bomo', 24)trans.commit()except:trans.rollback()raise3. session
connection是一般使用數據庫的方式，sqlalchemy還提供了另一種操作數據庫的方式，通過session對象，session可以記錄和跟蹤數據的改變，在適當的時候提交，并且支持強大的ORM的功能，下面是基本使用
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker# 數據庫連接字符串 DB_CONNECT_STRING = 'sqlite:Users/zhengxiankai/Desktop/Document/db.sqlite'# 創建數據庫引擎,echo為True,會打印所有的sql語句 engine = create_engine(DB_CONNECT_STRING, echo=True)# 創建會話類 DB_Session = sessionmaker(bind=engine)# 創建會話對象 session = DB_Session()# dosomething with session# 用完記得關閉，也可以用with session.close()上面創建了一個session對象，接下來可以操作數據庫了，session也支持通過sql語句操作數據庫
session.execute('select * from User') session.execute("insert into User(name, age) values('bomo', 13)") session.execute("insert into User(name, age) values(:name, :age)", {'name': 'bomo', 'age':12})# 如果是增刪改，需要commit session.commit()注意參數使用dict，并在sql語句中使用:key占位
4. ORM
上面簡單介紹了sql的簡單用法，既然是ORM框架，我們先定義兩個模型類User和Role，sqlalchemy的模型類繼承自一個由declarative_base()方法生成的類，我們先定義一個模塊Models.py生成Base類
from sqlalchemy.ext.declarative import declarative_baseBase = declarative_base()User.py
from sqlalchemy import Column, Integer, String from Models import Baseclass User(Base):__tablename__ = 'User'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String(50))age = Column('age', Integer)Role.py
from sqlalchemy import Column, Integer, String from Models import Baseclass Role(Base):__tablename__ = 'Role'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String(50))從上面很容易看出來，這里的模型對應數據庫中的表，模型支持的類型有Integer,?String,?Boolean,?Date,?DateTime,?Float，更多類型包括類型對應的Python的類型參見：這里
Column構造函數相關設置
-
name：名稱
-
type_：列類型
-
autoincrement：自增
-
default：默認值
-
index：索引
-
nullable：可空
-
primary_key：外鍵
更多介紹參見這里
接下來通過session進行增刪改查
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from User import User from Role import Role from Models import BaseDB_CONNECT_STRING = 'sqlite:Users/zhengxiankai/Desktop/Document/db.sqlite' engine = create_engine(DB_CONNECT_STRING, echo=True) DB_Session = sessionmaker(bind=engine) session = DB_Session()# 1. 創建表（如果表已經存在，則不會創建） Base.metadata.create_all(engine)# 2. 插入數據 u = User(name = 'tobi', age = 200) r = Role(name = 'user')# 2.1 使用add，如果已經存在，會報錯 session.add(u) session.add(r) session.commit() print r.id# 3 修改數據 # 3.1 使用merge方法，如果存在則修改，如果不存在則插入（只判斷主鍵，不判斷unique列） r.name = 'admin' session.merge(r)# 3.2 也可以通過這種方式修改 session.query(Role).filter(Role.id == 1).update({'name': 'admin'})# 4. 刪除數據 session.query(Role).filter(Role.id == 1).delete()# 5. 查詢數據 # 5.1 返回結果集的第二項 user = session.query(User).get(2)# 5.2 返回結果集中的第2-3項 users = session.query(User)[1:3]# 5.3 查詢條件 user = session.query(User).filter(User.id < 6).first()# 5.4 排序 users = session.query(User).order_by(User.name)# 5.5 降序（需要導入desc方法） from sqlalchemy import desc users = session.query(User).order_by(desc(User.name))# 5.6 只查詢部分屬性 users = session.query(User.name).order_by(desc(User.name)) for user in users:print user.name# 5.7 給結果集的列取別名 users = session.query(User.name.label('user_name')).all() for user in users:print user.user_name# 5.8 去重查詢（需要導入distinct方法） from sqlalchemy import distinct users = session.query(distinct(User.name).label('name')).all()# 5.9 統計查詢 #需要導入 from sqlalchemy.sql.expression import func user_count = session.query(User.name).order_by(User.name).count() age_avg = session.query(func.avg(User.age)).first() age_sum = session.query(func.sum(User.age)).first()# 5.10 分組查詢 users = session.query(func.count(User.name).label('count'), User.age).group_by(User.age) for user in users:print 'age:{0}, count:{1}'.format(user.age, user.count)# 5.11 # 創建Query查詢，filter是where條件，最后調用one()返回唯一行，如果調用all()則返回所有行: user = session.query(User).filter(User.id=='5').one()# 6.1 exists查詢(不存在則為~exists()) from sqlalchemy.sql import exists session.query(User.name).filter(~exists().where(User.role_id == Role.id)) # SELECT name AS users_name FROM users WHERE NOT EXISTS (SELECT * FROM roles WHERE users.role_id = roles.id)# 6.2 除了exists，any也可以表示EXISTS session.query(Role).filter(Role.users.any())# 7 random from sqlalchemy.sql.functions import random user = session.query(User).order_by(random()).first()session.close()5. 多表關系
上面的所有操作都是基于單個表的操作，下面是多表以及關系的使用，我們修改上面兩個表，添加外鍵關聯（一對多和多對一）
User模型
from sqlalchemy import Column, Integer, String from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship from Models import Baseclass User(Base):__tablename__ = 'users'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String(50))age = Column('age', Integer)# 添加角色id外鍵(關聯到Role表的id屬性)role_id = Column('role_id', Integer, ForeignKey('roles.id'))# 添加同表外鍵second_role_id = Column('second_role_id', Integer, ForeignKey('roles.id'))# 添加關系屬性，關聯到role_id外鍵上role = relationship('Role', foreign_keys='User.role_id', backref='User_role_id')# 添加關系屬性，關聯到second_role_id外鍵上second_role = relationship('Role', foreign_keys='User.second_role_id', backref='User_second_role_id')Role模型
from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship from Models import Baseclass Role(Base):__tablename__ = 'roles'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String(50))# 添加關系屬性，關聯到User.role_id屬性上users = relationship("User", foreign_keys='User.role_id', backref="Role_users")# 添加關系屬性，關聯到User.second_role_id屬性上second_users = relationship("User", foreign_keys='User.second_role_id', backref="Role_second_users")這里有一點需要注意的是，設置外鍵的時候ForeignKey('roles.id')這里面使用的是表名和表列，在設置關聯屬性的時候relationship('Role', foreign_keys='User.role_id', backref='User_role_id')，這里的foreign_keys使用的時候類名和屬性名
接下來就可以使用了
u = User(name='tobi', age=200)r1 = Role(name='admin') r2 = Role(name='user')u.role = r1 u.second_role = r2session.add(u) session.commit()# 查詢（對于外鍵關聯的關系屬性可以直接訪問，在需要用到的時候session會到數據庫查詢） roles = session.query(Role).all() for role in roles:print 'role:{0} users'for user in role.users:print '\t{0}'.format(user.name)print 'role:{0} second_users'for user in role.second_users:print '\t{0}'.format(user.name)上面表示的是一對多（多對一）的關系，還有一對一，多對多，如果要表示一對一的關系，在定義relationship的時候設置uselist為False（默認為True），如在Role中
class Role(Base):...user = relationship("User", uselist=False, foreign_keys='User.role_id', backref="Role_user")6. 多表查詢
由于關系數據庫的多個表還可以用外鍵實現一對多、多對多等關聯，相應地，ORM框架也可以提供兩個對象之間的一對多、多對多等功能。
例如，如果一個User擁有多個Book，就可以定義一對多關系如下：
class User(Base):__tablename__ = 'user'id = Column(String(20), primary_key=True)name = Column(String(20))# 一對多:books = relationship('Book')class Book(Base):__tablename__ = 'book'id = Column(String(20), primary_key=True)name = Column(String(20))# “多”的一方的book表是通過外鍵關聯到user表的:user_id = Column(String(20), ForeignKey('user.id'))?當我們查詢一個User對象時，該對象的books屬性將返回一個包含若干個Book對象的list。
?
多表查詢通常使用join進行表連接，第一個參數為表名，第二個參數為條件，例如
users = db.session.query(User).join(Role, Role.id == User.role_id)for u in users:print u.namejoin為內連接，還有左連接outerjoin，用法與join類似，右連接和全外鏈接在1.0版本上不支持，通常來說有這兩個結合查詢的方法基本夠用了，1.1版本貌似添加了右連接和全外連接的支持，但是目前只是預覽版
還可以直接查詢多個表，如下
result = db.session.query(User, Role).filter(User.role_id = Role.id) # 這里選擇的是兩個表，使用元組獲取數據 for u, r in result:print u.name?
數據庫遷移
sqlalchemy的數據庫遷移/升級有兩個庫支持alembic和sqlalchemy-migrate
由于sqlalchemy-migrate在2011年發布了0.7.2版本后，就已經停止更新了，并且已經不維護了，也積累了很多bug，而alembic是較后來才出現，而且是sqlalchemy的作者開發的，有良好的社區支持，所以在這里只學習alembic這個庫
alembic實現了類似git/svn的版本管理的控制，我們可以通過alembic維護每次升級數據庫的版本
1. 安裝
通過pip安裝，pip會自動安裝相關的依賴
pip install alembic2. 初始化
安裝完成后再項目根目錄運行
alembic init YOUR_ALEMBIC_DIRalembic會在根目錄創建YOUR_ALEMBIC_DIR目錄和alembic.ini文件，如下
yourproject/alembic.iniYOUR_ALEMBIC_DIR/env.pyREADMEscript.py.makoversions/3512b954651e_add_account.py2b1ae634e5cd_add_order_id.py3adcc9a56557_rename_username_field.py其中
-
alembic.ini?提供了一些基本的配置
-
env.py?每次執行Alembic都會加載這個模塊，主要提供項目Sqlalchemy Model 的連接
-
script.py.mako?遷移腳本生成模版
-
versions?存放生成的遷移腳本目錄
默認情況下創建的是基于單個數據庫的，如果需要支持多個數據庫或其他，可以通過alembic list_templates查看支持的模板
$ alembic list_templates Available templates:generic - Generic single-database configuration. multidb - Rudimentary multi-database configuration. pylons - Configuration that reads from a Pylons project environment.Templates are used via the 'init' command, e.g.:alembic init --template generic ./scripts3. 配置
使用之前，需要配置一下鏈接字符串，打開alembic.ini文件，設置sqlalchemy.url連接字符串，例如
sqlalchemy.url = sqlite:Users/zhengxiankai/Desktop/database.db其他參數可以參見官網說明：http://alembic.zzzcomputing.com/en/latest/tutorial.html
4. 創建數據庫版本
接下來我們創建一個數據庫版本，并新建兩個表
alembic revision -m 'create table'創建一個版本（會在yourproject/YOUR_ALEMBIC_DIR/versions/文件夾中創建一個python文件1a8a0d799b33_create_table.py）
該python模塊包含upgrade和downgrade兩個方法，在這里添加一些新增表的邏輯
"""create tableRevision ID: 4fd533a56b34 Revises: Create Date: 2016-09-18 17:20:27.667100""" from alembic import op import sqlalchemy as sa# revision identifiers, used by Alembic. revision = '4fd533a56b34' down_revision = None branch_labels = None depends_on = Nonedef upgrade():# 添加表op.create_table('account',sa.Column('id', sa.Integer, primary_key=True),sa.Column('name', sa.String(50), nullable=False),sa.Column('description', sa.Unicode(200)),)# 添加列# op.add_column('account', sa.Column('last_transaction_date', sa.DateTime))def downgrade():# 刪除表op.drop_table('account')# 刪除列# op.drop_column('account', 'last_transaction_date')這里使用到了了op對象，關于op對象的更多API使用，參見這里
這里生成的文件名是依照在alembic.ini文件聲明的模板來的，默認為版本號+名字，可以加上一些日期信息，否則不好排序，更多參數參見這里
file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(hour).2d_%%(minute).2d_%%(rev)s_%%(slug)s另外通常我們也改一下生成模板script.py.mako，加上編碼信息，否則在升級腳本中如果有中文會報錯
#!/usr/bin/python # -*- coding:utf-8 -*-5. 升級數據庫
剛剛實現了升級和降級的方法，通過下面命令升級數據庫到最新版本
alembic upgrade head這時候可以看到數據庫多了兩個表alembic_version和account，alembic_version存放數據庫版本
關于升級和降級的其他命令還有下面這些
# 升到最高版本 $ alembic upgrade head# 降到最初版本 $ alembic downgrade base# 升兩級 $ alembic upgrade +2# 降一級 $ alembic downgrade -1# 升級到制定版本 $ alembic upgrade e93b8d488143# 查看當前版本 $ alembic current# 查看歷史版本詳情 $ alembic history --verbose# 查看歷史版本（-r參數）類似切片 $ alembic history -r1975ea:ae1027 $ alembic history -r-3:current $ alembic history -r1975ea:6. 通過元數據升級數據庫
上面我們是通過API升級和降級，我們也可以直接通過元數據更新數據庫，也就是自動生成升級代碼，先定義兩個Model（User,?Role），這里我定義成三個文件
yourproject/YOUR_ALEMBIC_DIR/tutorial/DbModels.pyUser.pyRole.py代碼就放在一起了
from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String Base = declarative_base()class User(Base):__tablename__ = 'users'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String)class Role(Base):__tablename__ = 'roles'id = Column('id', Integer, primary_key=True, autoincrement=True)name = Column('name', String)在YOUR_ALEMBIC_DIR/env.py配置元數據
target_metadata = None改為
import os import sys# 這里需要添加相對路徑到sys.path，否則會引用失敗，嘗試過使用相對路徑，但各種不好使，還是使用這種方法靠譜些 sys.path.append(os.path.abspath(os.path.join(os.getcwd(), "../yourproject/tutorial/Db")))from User import User from Role import Role from Models import Base target_metadata = Base.metadataos.path.join(os.getcwd()這個獲取到的地址不是env.py的路徑，而是根目錄
在創建數據庫版本的時候添加--autogenerate參數，就會從Base.metadata元數據中生成腳本
alembic revision --autogenerate -m "add user table"這時候會在生成升級代碼
"""add user tableRevision ID: 97de1533584a Revises: 8678ab6d48c1 Create Date: 2016-09-19 21:58:00.758410""" from alembic import op import sqlalchemy as sa# revision identifiers, used by Alembic. revision = '97de1533584a' down_revision = '8678ab6d48c1' branch_labels = None depends_on = Nonedef upgrade():### commands auto generated by Alembic - please adjust! ###op.create_table('roles',sa.Column('id', sa.Integer(), nullable=False),sa.Column('name', sa.String(), nullable=True),sa.PrimaryKeyConstraint('id'))op.create_table('users',sa.Column('id', sa.Integer(), nullable=False),sa.Column('name', sa.String(), nullable=True),sa.PrimaryKeyConstraint('id'))op.drop_table('account')### end Alembic commands ###def downgrade():### commands auto generated by Alembic - please adjust! ###op.create_table('account',sa.Column('id', sa.INTEGER(), nullable=False),sa.Column('name', sa.VARCHAR(length=50), nullable=False),sa.Column('description', sa.VARCHAR(length=200), nullable=True),sa.Column('last_transaction_date', sa.DATETIME(), nullable=True),sa.PrimaryKeyConstraint('id'))op.drop_table('users')op.drop_table('roles')### end Alembic commands ###由于我沒有定義account模型，會被識別為刪除，如果刪除了model的列的聲明，則會被識別為刪除列，自動生成的版本我們也可以自己修改，然后執行升級命令即可升級alembic upgrade head
需要注意的是
Base.metadata聲明的類必須以數據庫中的一一對應，如果數據庫中有的表，而在元數據中沒有，會識別成刪除表
revision創建版本之前執行之前需要升級到最新版本
配置Base之前，需要保證所有的Model都已經執行（即導入）過一次了，否則無法讀取到，也就是需要把所有Model都import進來
數據庫升級有風險，升級前最好先檢查一遍upgrade函數，可以的話做好備份哈
?
常見問題
1. String長度問題
如果使用mysql數據庫，String類型對應的是VARCHAR類型，需要指定長度，否則會報下面錯誤，而在sqlite不會出現
(in table 'user', column 'name'): VARCHAR requires a length on dialect mysql?
參考鏈接
-
Auto Generating Migrations
-
tutorial
總結
以上是生活随笔為你收集整理的SQLAlchemy 教程 —— 进阶篇的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： Python base64编码解码
- 下一篇： docker将容器设置与宿主机为同一局域