python pandas dataFrame sqlAlchemy案例

python pandas、dataFramefrom sqlalchemy import create_engine,Engine,text,MetaData,Column,Table,Integer,String import pandas as pd #这是 导入 pandas 库并给它起了一个别名 pd。 from pandas import DataFrame from sqlalchemy.exc import OperationalError,ProgrammingError # 创建一次全局共享模拟 config 中的数据库配置 # mysqlpymysql://用户:密码主机:端口/数据库名 engine: Engine create_engine( mysqlpymysql://root:root127.0.0.1:3306/fastapi, pool_size5, # 连接池大小 max_overflow10, # 超出 pool_size 后最多再创建 10 个连接 pool_pre_pingTrue, # 每次从池取连接前先 ping 一下避免拿到断开的连接 echoTrue, # 设为 True 可打印所有 SQL方便调试 ) pandas 是 Python 最流行的数据处理库专门用来处理表格型数据类似 Excel 表格。 as pd 是别名之后用 pd 代替 pandas 写起来更短 DataFrame 是什么 DataFrame 表格可以理解为 Excel 中的一个 sheet 或数据库中的一张表。 id comment user_name like_count 1 张三的评论 张三 2 2 李四的评论 李四 5 3 王五的评论 王五 0 df.columns # 获取列名列表 df.to_dict() # 转为字典 df.to_sql() # 写入数据库 pd.read_sql() # 从数据库读取 df.head() # 看前几行 pandas.dataFrame.to_sql(table_name,engine,if_exists) # 写入数据库 table_name 目标数据库表名 字符串 engine SQLAlchemy 引擎 连接数据库的 engine 对象 if_exists 表已存在时的行为 fail / replace / append index 是否把 DataFrame 的行索引也写入数据库 True / False if_exists 三种模式 fail 表已存在 → 抛出异常 创建新表时用表存在说明有冲突 replace 表已存在 → 删掉重建再写入数据 覆盖全量数据 append 表已存在 → 追加数据 增量写入 三种取值完整说明 1. if_existsfail默认值 逻辑如果数据表已存在直接抛出异常 ValueError不执行任何写入操作 适用场景确认数据库不存在该表才写入防止误覆盖旧数据 示例报错ValueError: Table xxx already exists. 2. if_existsreplace 逻辑先删除原有整张表重新创建空表再写入当前 DataFrame 全部数据 特点 原表结构、索引、约束全部清空重建 表结构完全由当前 df 的列名、数据类型决定 适用场景全量更新每次覆盖整张表数据 3. if_existsappend 逻辑保留原有表结构和历史数据仅把当前 DataFrame 的数据追加到表末尾 强制要求否则报错 DataFrame 列名、列数量必须和数据库现有表完全一致 数据类型尽量匹配否则数据库会报类型错误 适用场景增量写入、日志流水、每日新增数据入库 def run_demo(): # 表的容器 meta MetaData() # Table()包裹表结构 table1 Table( t_department, meta, # 表名 所属表容器 Meta # 下面是字段定义 Column(id, Integer, primary_keyTrue,autoincrementTrue), # 主键 Column(name, String(200), nullableTrue), Column(department, String(200), nullableTrue), Column(salary, Integer, nullableTrue), ) meta.create_all(engine,tables[table1]) #append 追加写入 # ---- 准备数据 ---- dataFrame pd.DataFrame({ name: [Alice, Bob, Charlie], department: [Engineering, Sales, Engineering], salary: [50000, 60000, 70000], }) #append 追加写入数据 #dataFrame.to_sql(namet_department, conengine, if_existsappend, indexFalse) #fail表存在则报错抛出ValueError错误 # try: # dataFrame.to_sql(namet_department, conengine, if_existsfail, indexFalse) # except ValueError as e: # #ValueError # print(表已存在,,type(e)) #ValueError #replace模式 先drop t_department 再见表, 字符串是text数字是bigint , indexTrue会多生成一列indexbigint但这个表没有主键 dataFrame.to_sql(namet_department, conengine, if_existsreplace, indexFalse) pandas读数据非dataFrame读数据 dataList pd.read_sql(select * from t_department, conengine) 返回值类型是DataFrame print(type(dataList)) dataFrame转python类型支持3种 DataFrame.to_dict(prient records / list / dict) records [{列名: 值}, ...] 每条记录一个字典 [{name: Alice, ...}, {name: Bob, ...}] list {列名: [值列表]} 每列一个列表 {name: [Alice, Bob], salary: [50000, 60000]} dict默认 {列名: {行索引: 值}} {name: {0: Alice, 1: Bob}} 最常用的是 orientrecords直接得到一个列表每个元素是一条记录的字典。 # 转为字典 #[{name: Alice, department: Engineering, salary: 50000}, {name: Bob, department: Sales, salary: 60000}, {name: Charlie, department: Engineering, salary: 70000}] data_dict dataList.to_dict(orientrecords) print(data_dict) if __name__ __main__: run_demo()