一、介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上，使用关系对象映射进行数据库操作，简言之便是：将类和对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

安装：pip3 install sqlalchemy

组成部分：

Engine，框架的引擎
Connection Pooling ，数据库连接池
Dialect，选择连接数据库的DB API种类
Schema/Types，架构和类型
SQL Exprression Language，SQL表达式语言

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

二、使用

2.1 创建数据库表

2.1.1 数据类型和数据库的对应关系

类型名	MySQL类型	python类型	描述
SmallInteger	smallint	int	取值范围较小，一般为16位
Integer	int	int	普通整数，一般32位
BigInteger	bigint	int/long	不限精度的整数
Float	float	float	浮点数
Numeric	decimal	decimal.Decimal	定点数
String	varchar	str	变长字符串
Text	tinytext	str	变长字符串，64K,216−1216−1 2^{16}-1216−1=65535bytes
Text(65536)	mediumtext	str	变长字符串，max16M,224−1224−1 2^{24}-1224−1=16777215bytes
Text(16777216)	longtext	str	变长字符串，max32G,232−1232−1 2^{32}-1232−1=4294967295bytes
LargeBinary	blob	str	二进制文件，64K
LargeBinary(65536)	mediumblob	str	二进制，max16M
LargeBinary(16777216)	longblob	str	二进制，max32G
PickleType	blob	任何python对象	自动使用Pickle序列化，只有blob
Unicode	varchar	unicode	变长字符串
UnicodeText	text	unicode	变长字符串，64K
Boolean	tinyint	bool	布尔值
Date	date	datetime.date	日期
Time	time	date.time	时间
DateTime	datetime	datetime.datetime	日期和时间
Interval	datetime	datetime.timedelta	时间间隔
Enum	enum	str	一组字符串

属性	描述	生效值
primary_key	主键	True
unique	键值唯一性	True
index	索引	True
nullable	空值	True
default	默认值	null

2.1.2 示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
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
from sqlalchemy.orm import relationshipBase = declarative_base()###################### 单表示例 #########################
class Users(Base):__tablename__ = 'users'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)age = Column(Integer, default=18)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', 'extra'),                # 普通索引)class Hosts(Base):__tablename__ = 'hosts'id = Column(Integer, primary_key=True)name = Column(String(32), index=True)ctime = Column(DateTime, default=datetime.datetime.now)###################### 一对多示例 #########################
class Hobby(Base):__tablename__ = 'hobby'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)hobby_id = Column(Integer, ForeignKey("hobby.id"))# 与生成表结构无关，仅用于查询方便hobby = relationship("Hobby", backref='pers')   # 第一个字段表示关联的表，backref 表示反向查询的字段名称###################### 多对多示例 #########################
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')  # secondary表示第三张表class Server(Base):__tablename__ = 'server'id = Column(Integer, primary_key=True, autoincrement=True)hostname = Column(String(64), unique=True, nullable=False)def init_db():"""根据类创建数据库表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",max_overflow=0,   # 超过连接池大小外最多创建的连接pool_size=5,      # 连接池大小pool_timeout=30,  # 池中没有线程最多等待的时间，否则报错pool_recycle=-1   # 多久之后对线程池中的线程进行一次连接的回收（重置）)Base.metadata.create_all(engine)def drop_db():"""根据类删除数据库表"""engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?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()

指定关联列：hobby = relationship("Hobby", backref='pers',foreign_keys="Person.hobby_id")

2.2 操作数据库表

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()  # 每次执行数据库操作时，都需要创建一个session# ############# 执行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)session.commit()  # 提交事务
session.close()   # 关闭session

基于scoped_session实现线程安全

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
from models import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = scoped_session(Session)
"""
# 线程安全，基于本地线程实现每个线程用同一个session
# 特殊的：scoped_session中有原来方法的Session中的以下方法：public_methods = ('__contains__', '__iter__', 'add', 'add_all', 'begin', 'begin_nested','close', 'commit', 'connection', 'delete', 'execute', 'expire','expire_all', 'expunge', 'expunge_all', 'flush', 'get_bind','is_modified', 'bulk_save_objects', 'bulk_insert_mappings','bulk_update_mappings','merge', 'query', 'refresh', 'rollback','scalar'
)
"""# ############# 执行ORM操作 #############
obj1 = Users(name="alex1")
session.add(obj1)session.commit()  # 提交事务
session.close()   # 关闭session

多线程执行示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom 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 db import Usersengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)def task(arg):# 每个线程创建一个sessionsession = Session()obj1 = Users(name="alex1")session.add(obj1)session.commit()for i in range(10):t = threading.Thread(target=task, args=(i,))t.start()

2.3 基本增删改查示例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom 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 textfrom db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()# ################ 添加 ################
# 单个插入
obj1 = Users(name="fenglepeng")
session.add(obj1)
session.flush()
print(obj1.id)    # 输出新插入数据的主键，当只有一个主键时才能生效，如果设置两个及以上会取不到插入的数据# 批量插入
session.add_all([Users(name="wupeiqi"),Users(name="alex"),Hosts(name="c1.com"),
])
session.commit()   # 此时数据才插入到数据库中# ################ 删除 ################
session.query(Users).filter(Users.id > 2).delete()
session.commit()# ################ 修改 ################
# 使用 update
session.query(Users).filter(Users.id > 0).update({"name" : "099"})
session.query(Users).filter(Users.id > 0).update({Users.name: Users.name + "099"}, synchronize_session=False)       # 字符串
session.query(Users).filter(Users.id > 0).update({"age": Users.age + 1}, synchronize_session="evaluate")  # 数字# 或者直接修改查询到的对象的属性
obj1 = session.query(Users).filter(Users.id > 0).first()
obj1.name = 'fenglp'
session.commit()# ################ 查询 ################
r1 = session.query(Users).all()                               # 不加all，显示的是执行的sql
r2 = session.query(Users.name.label('xx'), Users.age).all()   # label 别名
r3 = session.query(Users).filter(Users.name == "alex").all()  # query表示查询的字段，filter表示过滤的字段
r4 = session.query(Users).filter_by(name='alex').all()        # 本表的字段
r5 = session.query(Users).filter_by(name='alex').first()# params 和Python中的format类似，用于替代前面的值
r6 = session.query(Users).filter(text("id<:value and name=:name")).params(value=224, name='fred').order_by(Users.id).all()
r7 = session.query(Users).from_statement(text("SELECT * FROM users where name=:name")).params(name='ed').all() # 嵌套查询，子查询session.close()

session 相关函数

db.session.add(obj)    # 添加
db.session.flush(obj)  # flush之后，可以取到obj的主键，但是当只有一个主键时才能生效，如果设置两个及以上会取不到插
db.session.rollback()  # 回滚
db.session.commit()    # 提交事务，commit执行了，但是此时sql层面的commit并没有执行
db.session.close()     # 关闭session

Model.query是db.session.query(Model)的快捷方式，它不可调用。如果您没有查询模型，请继续使用db.session.query(...)就像使用常规SQLAlchemy一样。

# 下面两种方式结果相同，第二种是第一种的简化版
db.session.query(Users).all() # 这种query中可以加参数
Users.query.all()             # 这种query中不能加参数

filter_by和filter的区别

filter_by表内部精确查询

User.query.filter_by(id=4).first()

filter 全局查询 id必须指明来源于那张表User，而且需要用等号，而不是赋值，建议使用filter

User.query.filter(User.id==4).first()

查询操作

################# 条件查询 ################
ret = session.query(Users).filter(Users.id = 1).all()   # 等于
ret = session.query(Users).filter(Users.id.between(1, 3), Users.name == 'eric').all()  # between
ret = session.query(Users).filter(Users.id.in_([1,3,4])).all()   # in
ret = session.query(Users).filter(~Users.id.in_([1,3,4])).all()  # 取反
ret = session.query(Users).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()  # 子查询################# 组合查询 ################
ret = session.query(Users).filter(Users.id > 1, Users.name == 'eric').all()  # 组合查询，默认是andfrom sqlalchemy import and_, or_
ret = session.query(Users).filter(and_(Users.id > 3, Users.name == 'eric')).all() # _and里面的关系是and
ret = session.query(Users).filter(or_(Users.id < 2, Users.name == 'eric')).all()
ret = session.query(Users).filter(or_(Users.id < 2,and_(Users.name == 'eric', Users.id > 3),Users.extra != "")).all()#################  通配符 % 表是多个，_ 表示一个 ################
ret = session.query(Users).filter(Users.name.like('e%')).all()
ret = session.query(Users).filter(~Users.name.like('e%')).all()################# 限制 ################
ret = session.query(Users).filter(Users.name.like('e%')).limit(10).all()  # offset
ret = session.query(Users).filter(Users.name.like('e%')).limit(10).offset(100).all()  # offset 和 limit
## 切片
ret = session.query(Users).filter(Users.name.like('e%')).slice(90,100).all()
ret = session.query(Users).filter(Users.name.like('e%'))[90:100]################# 排序 ################
ret = session.query(Users).order_by(Users.name.desc()).all()
ret = session.query(Users).order_by(Users.name.desc(), Users.id.asc()).all()################# 去重 ################
ret = session.query(Users).order_by(Users.name.desc()).distinct().all()################# 分组 ################
from sqlalchemy.sql import func################# 分组聚合 ################
ret = session.query(Users).group_by(Users.extra).all()
ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).all()
ret = session.query(func.max(Users.id),func.sum(Users.id),func.min(Users.id)).group_by(Users.name).having(func.min(Users.id) >2).all()################# 连表  ################
# 已经设置好外键,它自动找到关联的字段
ret = session.query(Users, Favor).filter(Users.id == Favor.nid).all()
ret = session.query(Person).join(Favor).all()                   # 默认是 inner join
ret = session.query(Person).join(Favor, isouter=True).all()     # isouter=True 表示左链接# 自己指定关联关系
ret = session.query(Person).join(Favor, Person.id == Favor.nid).all()               # 默认是 inner join
ret = session.query(Person).join(Favor, Person.id == Favor.nid, isouter=True).all() # isouter=True 表示左链接,或者也可以
ret = session.query(Person).outerjoin(Favor, Person.id == Favor.nid).all()          # 左连接# 返回值是一个列表，包含你查询的两个对象，如[<app.models.Favor object at 0x7f0a12692290>, <app.models.Person object at 0x7f0a12692290>]################# 组合 ################
q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union(q2).all()q1 = session.query(Users.name).filter(Users.id > 2)
q2 = session.query(Favor.caption).filter(Favor.nid < 2)
ret = q1.union_all(q2).all()

基于relationship操作ForeignKey

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom 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 db import Users, Hosts, Hobby, Personengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
########################## 添加 ############################
session.add_all([Hobby(caption='乒乓球'),Hobby(caption='羽毛球'),Person(name='张三', hobby_id=3),Person(name='李四', hobby_id=4),
])person = Person(name='张九', hobby=Hobby(caption='姑娘'))  # 同时创建，并自动添加关联
session.add(person)hb = Hobby(caption='人妖')
hb.pers = [Person(name='文飞'), Person(name='博雅')]
session.add(hb)
session.commit()########################## 查询 ############################
# 使用relationship正向查询
v = session.query(Person).first()
print(v.name)
print(v.hobby.caption)# 使用relationship反向查询
v = session.query(Hobby).first()
print(v.caption)
print(v.pers)session.close()

基于relationship操作m2m

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom 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 db import Users, Hosts, Hobby, Person, Group, Server, Server2Groupengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()
########################## 添加 ############################
session.add_all([Server(hostname='c1.com'),Server(hostname='c2.com'),Group(name='A组'),Group(name='B组'),
])
session.commit()s2g = Server2Group(server_id=1, group_id=1)
session.add(s2g)
session.commit()gp = Group(name='C组')
gp.servers = [Server(hostname='c3.com'),Server(hostname='c4.com')]
session.add(gp)
session.commit()ser = Server(hostname='c6.com')
ser.groups = [Group(name='F组'),Group(name='G组')]
session.add(ser)
session.commit()########################## 查询 ############################
# 使用relationship正向查询
v = session.query(Group).first()
print(v.name)
print(v.servers)# 使用relationship反向查询
v = session.query(Server).first()
print(v.hostname)
print(v.groups)session.close()

执行原生SQL语句

################# 第一种 ################
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1?charset=utf8",max_overflow=0,  # 超过连接池大小外最多创建的连接pool_size=5,     # 连接池大小pool_timeout=30, # 池中没有线程最多等待的时间，否则报错pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收（重置）
)def task(arg):conn = engine.raw_connection()cursor = conn.cursor()cursor.execute("select * from t1")result = cursor.fetchall()cursor.close()conn.close()for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第二种 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engineengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):conn = engine.contextual_connect()with conn:cur = conn.execute("select * from t1")result = cur.fetchall()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第三种 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threading
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.engine.base import Engine
from sqlalchemy.engine.result import ResultProxy
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=0, pool_size=5)def task(arg):cur = engine.execute("select * from t1")result = cur.fetchall()cur.close()print(result)for i in range(20):t = threading.Thread(target=task, args=(i,))t.start()################# 第四种 ################
#!/usr/bin/env python
# -*- coding:utf-8 -*-
import time
import threadingfrom 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 db import Users, Hostsengine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/s6", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)
session = Session()# 关联子查询
subqry = session.query(func.count(Server.id).label("sid")).filter(Server.id == Group.id).correlate(Group).as_scalar()
result = session.query(Group.name, subqry)
"""
SELECT `group`.name AS group_name, (SELECT count(server.id) AS sid
FROM server
WHERE server.id = `group`.id) AS anon_1
FROM `group`
"""# 查询
# cursor = session.execute('select * from users')
# result = cursor.fetchall()# 添加
cursor = session.execute('insert into users(name) values(:value)',params={"value":'wupeiqi'})
session.commit()
print(cursor.lastrowid)session.close()

注意: 查看连接 show status like 'Threads%';

从 datetime 字段中获取当日数据

from sqlalchemy import funcclass History(db.Model):__tablename__ = 'historys'id = db.Column(db.Integer, primary_key=True)date = db.Column(db.Date)datetime = db.Column(db.Datetime)# firstDay：某年某月的第一天，datetime类型
# lastDay：某年某月的最后一天，datetime类型
historys = History.query.filter(History.date.between(firstDay, lastDay)).all()
historys = History.query.filter(func.date(datetime).between(firstDay, lastDay)).all()# 格式化时间
func.date_format(History.datetime , "%Y-%m-%d %H:%i:%s")
# 将时间戳转化为时间
func.from_unixtime((Table.timestamp), "%Y-%m-%d %H:%i:%s")
# 获取12月份的所有数据
historys = History.query.filter(extract('month', History.date) == 12, extract('month', History.date) == 12).all()

动态增加条件

filter_by

filter_by用于查询简单的列名，不支持比较运算符。

filters = {’name': ‘fengyao', ‘age': 26}
User.query.filter_by(**filters).first()

filter

比filter_by的功能更强大，支持比较运算符，支持or_、in_等语法。

filters = {User.name == ‘fengyao’,User.age > 25
}
User.query.filter(*filters).first()

Flask 第三方组件之 SQLAlchemy相关推荐

Flask 第三方组件之 Migrate
flask-migrate是flask的一个扩展模块,主要是扩展数据库表结构的.类似于Django的python manage.py migrate 官方文档: http://flask-migrat ...
4.flask第三方组件
1.flask-session的使用在flask中,有一个app.session_interface = SecureCookieSessionInterface(),也就是存session,调用o ...
Flask 第三方组件之 WTForms
简介 WTForms是一个支持多个web框架的form组件,主要用于对用户请求数据进行验证. 安装: pip3 install wtforms 用户登录注册示例 1. 用户登录当用户登录时候,需要对 ...
Flask 第三方组件之 login
在了解使用Flask来实现用户认证之前,我们首先要明白用户认证的原理.假设现在我们自己去实现用户认证,需要做哪些事情呢? 首先,登录.用户能够输入用户名和密码进行登录,所以需要网页和表单,实现用户输入 ...
Flask 第三方组件之 script
Flask Script扩展提供向Flask插入外部脚本的功能,包括运行一个开发用的服务器,一个定制的Python shell,设置数据库的脚本,cronjobs,及其他运行在web应用之外的命令行任 ...
iOS 项目中用到的一些开源库和第三方组件
iOS 项目中用到的一些 iOS 开源库和第三方组件分享一下我目前所在公司 iOS 项目中用到的一些 iOS 开源库和第三方组件, 感谢开源, 减少了我们的劳动力, 节约了我们大量的时间, 让我们有 ...
React Native 项目常用第三方组件汇总
React Native 项目常用第三方组件汇总 https://www.jianshu.com/p/d9cd9a868764?utm_campaign=maleskine&utm_conte ...
android多线程下载原理,安卓多线程断点续传下载功能（靠谱第三方组件，原理demo）...
一,原生的DownloadManager 从Android 2.3(API level 9)开始,Android以Service的方式提供了全局的DownloadManager来系统级地优化处理长时间 ...
如何在Eclipse中查看Android源码或者第三方组件包源码
文章出处:http://blog.csdn.net/cjjky/article/details/6535426 在学习过程中如果经常阅读源码,理解程度会比较深,学习效率也会比较高,那么如何方便快捷的阅 ...

Flask 第三方组件之 SQLAlchemy

一、介绍

二、使用

2.1 创建数据库表

2.1.1 数据类型和数据库的对应关系

2.1.2 示例

2.2 操作数据库表

2.3 基本增删改查示例

session 相关函数

filter_by和filter的区别

查询操作

基于relationship操作ForeignKey

基于relationship操作m2m

执行原生SQL语句

从 datetime 字段中获取当日数据

动态增加条件

filter_by

filter

Flask 第三方组件之 SQLAlchemy相关推荐

最新文章

热门文章

Flask 第三方组件之 SQLAlchemy

一、介绍

二、 使用

2.1 创建数据库表

2.1.1 数据类型和数据库的对应关系

2.1.2 示例

2.2 操作数据库表

2.3 基本增删改查示例

session 相关函数

filter_by和filter的区别

查询操作

基于relationship操作ForeignKey

基于relationship操作m2m

执行原生SQL语句

从 datetime 字段中获取当日数据

动态增加条件

filter_by

filter

Flask 第三方组件之 SQLAlchemy相关推荐

最新文章

热门文章

二、使用