python pandas dataFrame sqlAlchemy案例

python pandas、dataFrame

from 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 中的数据库配置） # mysql+pymysql://用户:密码@主机:端口/数据库名 """ engine: Engine = create_engine( "mysql+pymysql://root:root@127.0.0.1:3306/fastapi", pool_size=5, # 连接池大小 max_overflow=10, # 超出 pool_size 后最多再创建 10 个连接 pool_pre_ping=True, # 每次从池取连接前先 ping 一下，避免拿到断开的连接 echo=True, # 设为 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_exists="fail"（默认值） 逻辑：如果数据表已存在，直接抛出异常 ValueError，不执行任何写入操作 适用场景：确认数据库不存在该表才写入，防止误覆盖旧数据 示例报错：ValueError: Table xxx already exists. 2. if_exists="replace" 逻辑：先删除原有整张表，重新创建空表，再写入当前 DataFrame 全部数据 特点： 原表结构、索引、约束全部清空重建 表结构完全由当前 df 的列名、数据类型决定 适用场景：全量更新，每次覆盖整张表数据 3. if_exists="append" 逻辑：保留原有表结构和历史数据，仅把当前 DataFrame 的数据追加到表末尾 强制要求（否则报错）： DataFrame 列名、列数量必须和数据库现有表完全一致 数据类型尽量匹配，否则数据库会报类型错误 适用场景：增量写入、日志流水、每日新增数据入库 """ def run_demo(): """# 表的容器""" meta = MetaData() """# Table()包裹表结构""" table1 = Table( "t_department", meta, # 表名 + 所属表容器 Meta # 下面是字段定义 Column("id", Integer, primary_key=True,autoincrement=True), # 主键 Column("name", String(200), nullable=True), Column("department", String(200), nullable=True), Column("salary", Integer, nullable=True), ) 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(name="t_department", con=engine, if_exists="append", index=False) #fail表存在则报错，抛出ValueError错误 # try: # dataFrame.to_sql(name="t_department", con=engine, if_exists="fail", index=False) # except ValueError as e: # #ValueError # print("表已存在,",type(e)) #ValueError """ """#replace模式 ，先drop t_department ，再见表, 字符串是text，数字是bigint , index=True会多生成一列index（bigint），但这个表没有主键""" dataFrame.to_sql(name="t_department", con=engine, if_exists="replace", index=False) """pandas读数据，非dataFrame读数据""" dataList =pd.read_sql("select * from t_department", con=engine) """返回值类型是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"}} 最常用的是 orient="records"，直接得到一个列表，每个元素是一条记录的字典。 """ """ # 转为字典 #[{'name': 'Alice', 'department': 'Engineering', 'salary': 50000}, {'name': 'Bob', 'department': 'Sales', 'salary': 60000}, {'name': 'Charlie', 'department': 'Engineering', 'salary': 70000}] """ data_dict = dataList.to_dict(orient="records") print(data_dict) if __name__ == '__main__': run_demo()