目录
SQLAlchemy简明教程
SQLAlchemy是Python中常用的一个ORM，SQLAlchemy分成三部分：
- ORM，就是我们用类来表示数据库schema的那部分
- SQLAlchemy Core，就是一些基础的操作，例如 update , insert 等等，也可以直接使用这部分来进行操作，但是它们写起来没有ORM那么自然
- DBAPI，这部分就是数据库驱动
它们的关系如下(图片来自官网)：
[
]
先来看看一个简单的例子：
import contextlib
from datetime import datetime
from sqlalchemy import create_engine, Column, Integer, DateTime, String
from sqlalchemy.orm import sessionmaker, declarative_base
# 创建 SQLAlchemy 引擎
engine = create_engine(
# 配置数据库地址：数据库类型+数据库驱动名称://用户名:密码@机器地址:端口号/数据库名
"mysql+pymysql://root:mysql_EEnSPA@localhost:3306/test",
echo=True, # 是不是要把所执行的SQL打印出来，一般用于调试
pool_size=16, # 连接池大小
max_overflow=32, # 连接池最大的大小
pool_recycle=3600 # 多久时间主动回收连接，
)
# 创建一个DbSession 类
# 每个实例DbSession都会是一个数据库会话。当然该类本身还不是数据库会话。
# 但是一旦我们创建了一个DbSession类的实例，这个实例将是实际的数据库会话。
DbSession = sessionmaker(autocommit=False, autoflush=False, bind=engine)
# 我们命名它是DbSession为了将它与我们从 SQLAlchemy 导入的Session区别开来。
# 稍后我们将使用Session（从 SQLAlchemy 导入的那个）
# 现在我们将使用declarative_base()返回一个类
# 要创建SessionLocal类，请使用函数sessionmaker：
# 稍后我们将用这个类继承，来创建每个数据库模型或类（ORM 模型）：
Base = declarative_base()
# 可以自动创建对应的表,一般是在创建表的时候使用
# Base.metadata.create_all(engine)
# 生成一个上下管理器方便使用
@contextlib.contextmanager
def get_session():
s = DbSession()
try:
yield s
s.commit()
except Exception as e:
s.rollback()
raise e
finally:
s.close()
class BaseMixin:
"""model的基类,所有model都必须继承"""
id = Column(Integer, primary_key=True)
created_at = Column(DateTime, nullable=False, default=datetime.now)
updated_at = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now,
index=True)
deleted_at = Column(DateTime) # 可以为空, 如果非空, 则为软删
class User(Base, BaseMixin):
__tablename__ = "user"
name = Column(String(36), nullable=False)
phone = Column(String(36), nullable=False, unique=True)
# 创建相应的表 ，如果表不存在，需要创建对应的表
# Base.metadata.create_all(engine)
注意上面的几点：
-
pool_recycle: 设置主动回收连接的时长，如果不设置，那么可能会遇到数据库主动断开连接的问题，例如
MySQL通常会为连接设置最大生命周期为八小时，如果没有通信，那么就会断开连接。因此不设置此选项可能就会遇到MySQL has gone away的报错。 -
engine:
engine是SQLAlchemy中位于数据库驱动之上的一个抽象概念，它适配了各种数据库驱动，提供了连接池等功能。其用法就是 如上面例子中，engine = create_engine(<数据库连接串>)，数据库连接串的格式是dialect+driver://username:password@host:port/database?参数这样的，dialect 可以是mysql,postgresql,oracle,mssql,sqlite，后面的 driver 是驱动，比如MySQL的驱动pymysql， 如果不填写，就使用默认驱动。再往后就是用户名、密码、地址、端口、数据库、连接参数了，来看几个例子： -
MySQL:
engine = create_engine('mysql+pymysql://scott:tiger@localhost/foo?charset=utf8mb4') -
PostgreSQL:
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase') -
Oracle:
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname') -
MS SQL:
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname') -
SQLite:
engine = create_engine('sqlite:////absolute/path/to/foo.db') -
Session:
Session的意思就是会话，也就是说，是一个逻辑组织的概念，因此，这需要靠你的业务逻辑来划分哪些操作使用同一个Session， 哪些操作又划分为不同的业务操作，详见 这里。 举个简单的例子，以web应用为例，一个请求里共用一个Session就是一个好的例子，一个异步任务执行过程中使用一个Session也是一个例子。 但是注意，不能直接使用Session，而是使用Session的实例，借助上面的代码，我们可以直接这样写：with get_session() as s: print(s.query(User).first()) -
Base:
Base是ORM中的一个基类，通过集成Base，我们才能方便的使用一些基本的查询，例如s.query(User).filter_by(User.name="nick").first()。 -
BaseMixin:
BaseMixin是定义的一些通用的表结构，通过Mixin的方式集成到类里，比如上面的定义，我们常见的表结构里，都会有 ID、创建时间、更新时间，软删除标志 等等，我们把它作为一个独立的类，这样通过继承即可获得相关表属性，省得重复写多次。
表的设计¶
表的设计通常就如 User 表一样：
class User(Base, BaseMixin):
__tablename__ = "user"
Name = Column(String(36), nullable=False)
Phone = Column(String(36), nullable=False, unique=True)
首先使用 tablename 自定义表名，接着写各个表中的属性，也就是对应在数据库表中的列(column)，常见的类型有：
>$ egrep '^class ' ~/.pyenv/versions/3.6.0/lib/python3.6/site-packages/sqlalchemy/sql/sqltypes.py
class _LookupExpressionAdapter(object):
class Concatenable(object):
class Indexable(object):
class String(Concatenable, TypeEngine):
class Text(String):
class Unicode(String):
class UnicodeText(Text):
class Integer(_LookupExpressionAdapter, TypeEngine):
class SmallInteger(Integer):
class BigInteger(Integer):
class Numeric(_LookupExpressionAdapter, TypeEngine):
class Float(Numeric):
class DateTime(_LookupExpressionAdapter, TypeEngine):
class Date(_LookupExpressionAdapter, TypeEngine):
class Time(_LookupExpressionAdapter, TypeEngine):
class _Binary(TypeEngine):
class LargeBinary(_Binary):
class Binary(LargeBinary):
class SchemaType(SchemaEventTarget):
class Enum(Emulated, String, SchemaType):
class PickleType(TypeDecorator):
class Boolean(Emulated, TypeEngine, SchemaType):
class _AbstractInterval(_LookupExpressionAdapter, TypeEngine):
class Interval(Emulated, _AbstractInterval, TypeDecorator):
class JSON(Indexable, TypeEngine):
class ARRAY(SchemaEventTarget, Indexable, Concatenable, TypeEngine):
class REAL(Float):
class FLOAT(Float):
class NUMERIC(Numeric):
class DECIMAL(Numeric):
class INTEGER(Integer):
class SMALLINT(SmallInteger):
class BIGINT(BigInteger):
class TIMESTAMP(DateTime):
class DATETIME(DateTime):
class DATE(Date):
class TIME(Time):
class TEXT(Text):
class CLOB(Text):
class VARCHAR(String):
class NVARCHAR(Unicode):
class CHAR(String):
class NCHAR(Unicode):
class BLOB(LargeBinary):
class BINARY(_Binary):
class VARBINARY(_Binary):
class BOOLEAN(Boolean):
class NullType(TypeEngine):
class MatchType(Boolean):
常见操作¶
我们来看看使用SQLAlchemy完成常见的操作，例如增删查改：
常见查询操作¶
-
SELECT * FROM user应该这样写：with get_session() as s: print(s.query(User).all()) -
SELECT * FROM user WHERE name='nick'应该这样写：with get_session() as s: print(s.query(User).filter_by(name='nick').all()) # 需要注意，这里的name 是字段名称，必须和数据库一一匹配才可以 print(s.query(User).filter(User.name == 'nick').all()) # 这样写是等同效果的 -
SELECT * FROM user WHERE name='nick' LIMIT 1应该这样写：with get_session() as s: print(s.query(User).filter_by(name='nick').first())如果需要加判定，例如确保只有一条数据，那就把
first()替换为one()，如果确保一行或者没有，那就写one_or_none()。 -
SELECT * FROM user ORDER BY id DESC LIMIT 1应该这样写：with get_session() as s: print(s.query(User).order_by(User.id.desc()).first()) -
SELECT * FROM user ORDER BY id DESC LIMIT 1 OFFSET 20应该这样写：with get_session() as s: print(s.query(User).order_by(User.id.desc()).offset(20).first())
常见删除操作¶
-
DELETE FROM user应该这样写：with get_session() as s: s.query(User).delete() -
DELETE FROM user WHERE name='nick'应该这样写：with get_session() as s: s.query(User).filter_by(name='nick').delete() -
DELETE FROM user WHERE name='nick' LIMIT 1应该这样写：with get_session() as s: s.query(User).filter_by(name='nick').limit(1).delete()
常见更新操作¶
-
UPDATE user SET name='nick'应该这样写：with get_session() as s: s.query(User).update({'name': 'nick'}) -
UPDATE user SET name='nick' WHERE id=1应该这样写：with get_session() as s: s.query(User).filter_by(id=1).update({'name': 'nick'})也可以通过更改实例的属性，然后提交：
with get_session() as s: user = s.query(User).filter_by(User.id=1).one() user.name = 'nick' s.commit()
常见插入操作¶
这个就简单了，实例化对象，然后 session.add，最后提交：
with get_session() as s:
# 增加单个
user = User()
s.add(user)
# 增加多个
users = [User(name="lisi1", phone="12313321321"), User(name="wangwu1", phone="12313321331")]
s.add_all(users)
s.commit()
常见连表查询¶
SQLAlchemy 中可以直接使用 join语句 ：
with get_session() as s:
s.query(Customer).join(Invoice).filter(Invoice.amount == 8500)
可以是这么几种写法：
query.join(Address, User.id==Address.user_id) # explicit condition
query.join(User.addresses) # specify relationship from left to right
query.join(Address, User.addresses) # same, with explicit target
query.join('addresses') # same, using a string
数据库migration¶
我们使用alembic来做数据库migration，首先安装：
>$ pip install alembic
>$ alembic init alembic # 此处 alembic init 后接的是保存migration的文件夹名称
然后我们要修改 alembic/env.py (假设你设置的保存migration的文件夹名称就是 alembic)，将对应部分修改成如下：
config.set_main_option(
'sqlalchemy.url', config.SQLALCHEMY_DATABASE_URI
)
target_metadata = Base.metadata # 从任意一个我们的model可以拿到总的Base
engine = target_metadata.bind
因为SQLAlchemy会把表的信息存储在 metadata 里，而我们都继承了 Base ，因此可以 通过 Base.metadata 来拿到所有表的信息，这样子 alembic 才能够拿到表的结构，然后和数据库进行对比，生成 migration 脚本：
>$ alembic revision --autogenerate -m '本次migration的信息，相当于git提交时的评论'
这一篇中我们看了如何使用SQLAlchemy来进行常见的操作，我们首先从如何定义表开始，接着我们注意看了常见的SQL操作对应的 SQLAlchemy操作是怎样的，最后我们看了以下alembic应该怎么配置才能自动生成migration脚本。
参考资料：
标签：engine,SQLAlchemy,name,Python,User,query,class,user From： https://www.cnblogs.com/alex-oos/p/18398178