无从致书以观，每假借于藏书之家，手自笔录，计日以还。
ORM（object relational mapping） 对象关系映射关系 ，面向对象的对象模型和关系型数据之间的相互转换。
大白话就是基于类对象的方式操作数据库，一个类对应一张表，一个类对象对应一行数据。
SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
pip3 install sqlalchemy
Engine:框架的引擎
Connection Pooling:数据库连接池,用来管理应用程序到数据库的连接
Dialect:用来对接不同的数据库驱动，这些驱动要实现数据库的DB API:mysql,sqlite,oracle
Schema/Types: 1.实现了对SQL的DDL(Data Definition Language)的封装，其实就是实现的类对应一张表，对象对应一行数据，SQL Expression Language的操作对象就是这里定义的Table，2.同时也是架构和类型,数据库升降级就是是通过schema控制的（db sync）
SQL Expression Language:SQL表达式语言实现了对数据库的SELECT、DELETE、UPDATE等语句的封装，是实现ORM层的基础。
SQLAlchemy本身无法操作数据库，其必须用pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>
pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]
MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>
cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...]
更多：http://docs.sqlalchemy.org/en/latest/dialects/index.html
值得注意的是：
利用SQLAlchemy(部份功能)连接池功能结合原生SQL可以操作数据库
利用了SQLAlchemy 上层ORM结合原生SQL也可以操作数据库
直接利用ORM操作数据库（常规用法）
利用SQLAlchemy(部份功能)连接池功能结合原生SQL可以操作数据库，方法还有很多种，如：
eg1(不常用):
cur = engine.execute("select * from t1")
result = cur.fetchall()
eg2(不常用)
conn = engine.contextual_connect()
with conn:
cur = conn.execute("select * from t1")
result = cur.fetchall()
这里只列出一种来理解学习
import threading
import pymysql
from sqlalchemy import create_engine
engine = create_engine(
'mysql+pymysql://root:qq11@192.168.1.6:3306/db1?charset=utf8',
max_overflow=1,
pool_size=2,
pool_timeout=30,
pool_recycle=-1,
def task(arg):
conn = engine.raw_connection()
cur = conn.cursor(pymysql.cursors.DictCursor) # pymysql.cursors.DictCursor表示返回字典形式，默认元组形式
cur.execute('select * from userinfo')
# cur.execute('select sleep(2)')
result = cur.fetchall()
cur.close()
conn.close()
print(result)
for i in range(10):
t = threading.Thread(target=task,args=(i,))
t.start()
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
from sqlalchemy.sql import text
from sqlalchemy.engine.result import ResultProxy
from models import UserInfo # 前提是models.py定义了UserInfo表，models.py文件中导入UserInfo表
engine = create_engine('mysql+pymysql://root:qdf@192.168.1.6:3306/db1?charset=utf8', max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
cursor = session.execute('select * from userinfo')
result = cursor.fetchall()
print(result)
# cursor = session.execute('insert into userinfo(name,email) values(:name,:email)',params={"name":'xxrr','email':'xxrr@qq.com'})
# session.commit()
# print(cursor.lastrowid)
session.close()
创建数据库表
# models.py
import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index
Base = declarative_base()
## 单表创建
class UserInfo(Base):
__tablename__ = 'userinfo'
id = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=False)
email = Column(String(32), unique=True)
ctime = Column(DateTime, default=datetime.datetime.now)
# extra = Column(Text, nullable=True)
__table_args__ = (
# UniqueConstraint('id', 'name', name='uix_id_name'),
Index('ix_id_name', 'name', 'email'),
def init_db():
根据类创建数据库表
:return:
engine = create_engine(
'mysql+pymysql://root:qdf@192.168.1.6:3306/db1?charset=utf8',
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
Base.metadata.create_all(engine)
def drop_db():
根据类删除数据库表
:return:
engine = create_engine(
'mysql+pymysql://root:qwer1011@192.168.1.6:3306/db1?charset=utf8',
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
Base.metadata.drop_all(engine)
if __name__ == '__main__':
drop_db()
init_db()
# python models.py即可生成数据库，然后就可以利用ORM操作数据库了.
简单单表增删改查操作
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import UserInfo
engine = create_engine('mysql+pymysql://root:qfg@192.168.1.6:3306/db1?charset=utf8', max_overflow=0, pool_size=5)
SessionClass = sessionmaker(bind=engine)
# 每次执行数据库操作时，都需要创建一个session
session = SessionClass()
# ############# 添加 #############
obj1 = UserInfo(name="2p",email='2p@qq.com')
session.add(obj1)
# 提交事务
session.commit()
# ############# 查询 #############
result = session.query(UserInfo).all()
for row in result:
print(row.id,row.name,row.email)
# result = session.query(UserInfo).filter(UserInfo.id > 3)
# for row in result:
# print(row.id,row.name,row.email)
result = session.query(UserInfo.id.label('sn'),UserInfo.name,UserInfo.email).all()
for row in result:
print(row.sn,row.name,row.email)
# ############# 修改 #############
effect_row_num = session.query(UserInfo).filter(UserInfo.id > 3).update({'name':'new_name',})
session.commit()
print('update',effect_row_num)
# ############# 删除 #############
effect_row_num = session.query(UserInfo).filter(UserInfo.id > 3).delete()
session.commit()
print('del',effect_row_num)
# 关闭session
session.close()
包含M2M，FK的多表创建参考如下
# ##################### 一对多示例 #########################
class PersonType(Base):
__tablename__ = 'persontype'
id = Column(Integer, primary_key=True)
caption = Column(String(50), default='普通用户')
class Person(Base):
__tablename__ = 'person'
nid = Column(Integer, primary_key=True)
name = Column(String(32), index=True, nullable=True)
person_id = Column(Integer, ForeignKey("persontype.id"))
# 与生成表结构无关，仅用于查询方便
ptype = relationship("PersonType", backref='pers')
# ##################### 多对多示例 #########################
class Server2Group(Base):
__tablename__ = 'server2group'
id = Column(Integer, primary_key=True, autoincrement=True)
server_id = Column(Integer, ForeignKey('server.id'))
group_id = Column(Integer, ForeignKey('group.id'))
class Group(Base):
__tablename__ = 'group'
id = Column(Integer, primary_key=True)
name = Column(String(64), unique=True, nullable=False)
# 与生成表结构无关，仅用于查询方便
servers = relationship('Server', secondary='server2group', backref='groups')
class Server(Base):
__tablename__ = 'server'
id = Column(Integer, primary_key=True, autoincrement=True)
hostname = Column(String(64), unique=True, nullable=False)
四. 总结