sqlalchemy
是一个基于python的orm框架，可以让我们在python中可以使用sql操作数据库
flask 中没有orm框架 都是使用sqlalchemy作为操作数据库表的模块
fastapi 也是使用的sqlchemy
1.安装 pip3.8 install sqlalchemy
# sqlalchemy本身是无法操作数据库的，必须要使用pymysql一起
如何使用sqlalchemy写原生sql
# 第一步：导入
from sqlalchemy import create_engine
# 第二步：生成引擎对象
engine = create_engine(
"mysql+pymysql://moongod:[email protected]:3306/FlowProps",
# 连接mysql用户是moongod密码是123 连接的库名是FlowProps
max_overflow=2, # 超过连接池大小外最多创建的连接
pool_size=10, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 第三步：使用引擎获取连接，操作数据库
conn = engine.raw_connection()
cursor = conn.cursor()
cursor.execute('select * from app_props')
print(cursor.fetchall())
使用sqlalchemy创建表
# 第一步导入
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime
import datetime
# 第二步 生成基类
Base = declarative_base()
# 第三步 继承基类编写表 和 字段
class User(Base):
id = Column(Integer, primary_key=True，autoincrement=True)
name = Column(String(32), index=True, nullable=False)
# nullable=False 不可以为空
email = Column(String(36), unique=True)
ctime = Column(DateTime, default=datetime.datetime.now)
extra = Column(Text, nullable=True)
__tablename__ = 'users'
# 第四步 生成数据库连接 首先要自己创好库
engine = create_engine(
"mysql+pymysql://moongod:[email protected]:3306/text",
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 第五步 执行表迁移 创建表
Base.metadata.create_all(engine)
# 删除所有表的指令
Base.metadata.drop_all(engine)
参数：
primary_key=True 设置主键 主键自带非空唯一自增
Integer 整形
String(32) 字符串类型 最多32位
nullable=False 不可以为空
unique=True 唯一值索引
# 第七步执行表迁移
# 库需要自己去创建，举例text库
engine = create_engine(
"mysql+pymysql://root:[email protected]:3306/text",
# 用户root密码123
max_overflow=0, # 超过连接池大小外最多创建的连接
pool_size=5, # 连接池大小
pool_timeout=30, # 池中没有线程最多等待的时间，否则报错
pool_recycle=-1 # 多久之后对线程池中的线程进行一次连接的回收（重置）
)
# 把表同步到数据库 这样就直接创建表了（把被Base管理的所有表，都创建到数据库）
Base.metadata.create_all(engine)
# 把所有表删除
# Base.metadata.drop_all(engine)