1. 常用的sqlalchemy字段类型

类型名	python中类型	说明
Integer	int	普通整数，一般是32位
SmallInteger	int	取值范围小的整数，一般是16位
BigInteger	int或long	不限制精度的整数
Float	float	浮点数
Numeric	decimal.Decimal	普通整数，一般是32位
String	str	变长字符串
Text	str	变长字符串，对较长或不限长度的字符串做了优化
Unicode	unicode	变长Unicode字符串
UnicodeText	unicode	变长Unicode字符串，对较长或不限长度的字符串做了优化
Boolean	bool	布尔值
Date	datetime.date	时间
Time	datetime.datetime	日期和时间
LargeBinary	str	二进制文件

2. 常用的Column选项

选项名	说明
primary_key	如果为True，代表表的主键
unique	如果为True，代表这列不允许出现重复的值
index	如果为True，为这列创建索引，提高查询效率
nullable	如果为True，允许有空值，如果为False，不允许有空值
default	为这列定义默认值
comment	字段注释

3. 常用的关系选项

选项名	说明
backref	在关系的另一模型中添加反向引用
primaryjoin	明确指定两个模型之间使用的联结条件
uselist	如果为False，不使用列表，而使用标量值
order_by	指定关系中记录的排序方式
secondary	指定多对多中记录的排序方式

4. 简单操作

安装sqlalchemy模块

pip install sqlalchemy

4.1 连接数据库并创建表

# 导入模块
from sqlalchemy import create_engine, Column, Integer, String, DATETIME
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建对象的基类
Base = declarative_base()

# 定义表对象
class Users(Base):
   # 表的名字
    __tablename__ = "user"
  
    # 表的结构
    id = Column(Integer, primary_key=True)
    username = Column(String(64), unique=True, nullable=False, comment="用户名")
    password = Column(String(64), unique=True, nullable=False, comment="密码")

    def __init__(self, id, username, password, create_time, update_time):
        self.id = id
        self.username = username
        self.password = password


# 初始化数据库连接
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1/system_user?charset=utf8mb4", echo=True)
# 创建DBsession类型
DBSession = sessionmaker(bind=engine)
# 创建定义的表
Base.metadata.create_all(engine)

create_engine()用来初始化数据库连接。SQLAlchemy用一个字符串表示连接信息：

'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名'

Engine内部维护了一个Pool（连接池）和Dialect（方言），方言来识别具体连接数据库种类。

create_engine还有其它可选的参数，比如：

engine = create_engine("mysql+pymysql://root:123456@127.0.0.1/system_user?charset=utf8mb4", 
                       echo=True, 
                       pool_size=5,
                       pool_recycle=-1,
                       query_cache_size=500)
# echo=True:当设置为True时会将orm语句转化为sql语句打印，一般debug的时候可用
# pool_size:连接池的大小，默认为5个，设置为0时表示连接无限制
# pool_recycle:设置时间以限制数据库多久没连接自动断开,默认-1没有限制
# query_cache_size：用于缓存 SQL 字符串的缓存大小查询的形式。设置为零以禁用缓存。

4.2 插入数据

engine = create_engine("mysql+pymysql://root:123456@81.70.147.98/system_user?charset=utf8mb4", echo=True)
DBSession = sessionmaker(bind=engine)
session = DBSession()
# 定义添加的数据
user_add = Users(
    id=1,
    username="abc",
    password="123456"
)
# 利用session.add()向数据库添加
session.add(user_add)
session.commit()
session.close()

4.3 查询数据

user_message = session.query(Users).filter_by(id=1).all()
# user_message = session.query(Users).filter_by(Users.id == 1).all()
for i in user_message:
    print(i.username)
    
# 通过session.query()我们查询返回了一个Query对象，此时还没有去具体的数据库中查询，只有当执行具体的.all()，.first()等函数时才会真的去操作数据库。

filter	filter_by
支持所有比较运算符，相等比较用比较用==	只能使用"="，"!="和"><"
过滤用类名.属性名	过滤用属性名
不支持组合查询，只能连续调用filter变相实现	参数是**kwargs，支持组合查询
支持and，or和in等

4.4 更改数据

session.query(Users).filter_by(id=1).update({"username": "ccc"})
# 或者
user = session.query(Users).filter_by(id=1).first()
user.username = "aaa"
session.add(user)

4.5 删除数据

delete_user = session.query(Users).filter_by(id=1).first()
if delete_user:
    session.delete(delete_user)
    session.commit()
# 或者
session.query(Users).filter_by(id=1).delete()
session.commit()

5. 特殊操作

5.1 时间列自动更新

from sqlalchemy import create_engine, Column, Integer, String, TIMESTAMP, text

...
create_time = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP"), comment="创建时间", nullable=False)
update_time = Column(TIMESTAMP, server_default=text("CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP"), comment="更新时间", nullable=False)
...

5.2 批量添加

# 第一种方法
user_add_list = [
    Users(id=1, username="aaa", password="1", create_time=None, update_time=None),
    Users(id=2, username="bbb", password="2", create_time=None, update_time=None),
    Users(id=3, username="ccc", password="3", create_time=None, update_time=None),
    Users(id=4, username="ddd", password="4", create_time=None, update_time=None),
]
session.add_all(user_add_list)
session.commit()

# 第二种方法
user_add_list = [
    Users(id=1, username="aaa", password="1", create_time=None, update_time=None),
    Users(id=2, username="bbb", password="2", create_time=None, update_time=None),
    Users(id=3, username="ccc", password="3", create_time=None, update_time=None),
    Users(id=4, username="ddd", password="4", create_time=None, update_time=None),
]
session.bulk_save_objects(user_add_list)
session.commit()

5.3 查询参数

# 返回所有符合条件的记录
session.query(Users).filter_by(id=1).all()
# 返回符合条件的第一条记录
session.query(Users).filter_by(id=1).first()
# 返回一个对象，如果有多个报错
session.query(Users).filter_by(Users.id=1).one()
# 限制返回条数
session.query(Users).filter(Users.id=1).limit(2).all()
# 从第3条开始返回记录
session.query(Users).filter(Users.id=1).offset(2).all()
# 切片返回记录
session.query(Users).filter(Users.id=1).slice(2,3).all()
# 按顺序排序返回记录
session.query(Users).filter(Users.username="abc").order_by(id).all()
# 降序返回记录
session.query(Users).filter(Users.username="abc").order_by(desc(id)).all()
# 不等于
session.query(Users).filter(Users.id!=1).all()
# 模糊匹配
session.query(Users).filter(Users.username.like("aaa")).all()
session.query(Users).filter(Users.username.notlike("aaa")).all()
# 成员属于/不属于
session.query(User).filter(User.username.in_(['aaa','bbb'])).all()
session.query(User).filter(User.username.notin_(['aaa','bbb'])).all()
# 空判断
session.query(User).filter(User.username==None).all()
session.query(User).filter(User.username.is_(None)).all()
session.query(User).filter(User.username.isnot(None)).all()
# 多条件
session.query(User).filter(User.username.isnot(None),User.password=='123').all()
# and or
session.query(User).filter(or_(User.username=='aaa',User.password=='123')).all()
session.query(User).filter(and_(User.username=='bbb',User.password=='111')).all()
# 计算个数
session.query(Users).filter_by(id=1).count()

原文地址：https://www.cnblogs.com/os-linux/p/15079401.html