简介
SQLAlchemy 是python中，通过ORM操作数据库的框架。它简化了应用程序开发人员在原生SQL上的操作，使开发人员将主要精力都放在程序逻辑上，从而提高开发效率。它提供了一整套著名的企业级持久性模式，设计用于高效和高性能的数据库访问。
安装
pip install SQLAlchemy
连接数据（engine）
任何SQLAlchemy应用程序的开始都是一个Engine对象，此对象充当连接到特定数据库的中心源，提供被称为connection pool的对于这些数据库连接。
Engine对象通常是一个只为特定数据库服务器创建一次的全局对象，并使用一个URL字符串进行配置，该字符串将描述如何连接到数据库。
from sqlalchemy import create_engine
engine = create_engine('sqlite:///:memory:', echo=True)
create_engine的常用参数
- echo=False – 如果为真，引擎将记录所有语句以及
repr()其参数列表的默认日志处理程序。 - enable_from_linting – 默认为True。如果发现给定的SELECT语句与将导致笛卡尔积的元素取消链接，则将发出警告。
- encoding – 默认为
utf-8 - future – 使用2.0样式
- hide_parameters – 布尔值，当设置为True时，SQL语句参数将不会显示在信息日志中，也不会格式化为 StatementError 对象。
- listeners – 一个或多个列表
PoolListener将接收连接池事件的对象。 - logging_name – 字符串标识符，默认为对象id的十六进制字符串。
- max_identifier_length – 整数；重写方言确定的最大标识符长度。
- max_overflow=10 – 允许在连接池中“溢出”的连接数，即可以在池大小设置（默认为5）之上或之外打开的连接数。
- pool_size=5 – 在连接池中保持打开的连接数
- plugins – 要加载的插件名称的字符串列表。
数据库连接字符串格式：
dialect+driver://username:password@host:port/database
例如 MySQL：mysql+pymysql://username:password@host:port/database
postgresql：postgresql+psycopg2://username:password@host:port/database
声明映射
创建一个映射类，是基于基类定义的，每个映射类都要继承这个基类 declarative_base()。
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import declarative_base
Base = declarative_base()
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
fullname = Column(String)
nickname = Column(String)
def __repr__(self):
return "<User(name='%s', fullname='%s', nickname='%s')>" % (
self.name, self.fullname, self.nickname)
tablename 代表表名
Column : 代表数据表中的一列，内部定义了数据类型，如果字段名称与数据定义的字段名称不一致时，可以在Column中指定
id = Column('user_id',Integer, primary_key=True)
**primary_key：**主键
创建数据库
我们可以通过检查__table__属性查看关于表的信息，称为table metadata：
>>> User.__table__
Table('users', MetaData(),
Column('id', Integer(), table=<users>, primary_key=True, nullable=False),
Column('name', String(), table=<users>),
Column('fullname', String(), table=<users>),
Column('nickname', String(), table=<users>), schema=None)
创建表：
Base.metadata.create_all(engine)
# BEGIN...
# CREATE TABLE users (
# id INTEGER NOT NULL,
# name VARCHAR,
# fullname VARCHAR,
# nickname VARCHAR,
# PRIMARY KEY (id)
# )
# [...] ()
# COMMIT
数据操作
添加数据到数据库
创建一个User对象的实例，然后创建会话
from sqlalchemy.orm import sessionmaker
Session = sessionmaker(bind=engine)
session = Session() # 实例化会话
# 对象实例
user = User(name='ed', fullname='Ed Jones', nickname='edsnickname')
session.add(user)
session.commit() # 提交数据
查询数据
通过 query 关键字查询。
- query.filter() 过滤
- query.filter_by() 根据关键字过滤
- query.all() 返回列表
- query.first() 返回第一个元素
- query.one() 有且只有一个元素时才正确返回
- query.one_or_none()，类似one，但如果没有找到结果，则不会引发错误
- query.scalar()，调用one方法，并在成功时返回行的第一列
- query.count() 计数
- query.order_by() 排序
for instance in session.query(User).filter(User.id > 10).order_by(User.id):
print(instance.name, instance.fullname)
query.join()
连接查询
session.query(User).join(Address).filter(Address.email_address=='bob@foxmail.com'). all()
查询常用筛选器运算符
# 等于
query.filter(User.name == 'ed')
# 不等于
query.filter(User.name != 'ed')
# like和ilike
query.filter(User.name.like('%ed%'))
query.filter(User.name.ilike('%ed%')) # 不区分大小写
# in
query.filter(User.name.in_(['ed', 'wendy', 'jack']))
query.filter(User.name.in_(
session.query(User.name).filter(User.name.like('%ed%'))
))
# not in
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
# is
query.filter(User.name == None)
query.filter(User.name.is_(None))
# is not
query.filter(User.name != None)
query.filter(User.name.is_not(None))
# and
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
# or
from sqlalchemy import or_
query.filter(or_(User.name == 'ed', User.name == 'wendy'))
# match
query.filter(User.name.match('wendy'))
