pandas.DataFrame.to_sql#

DataFrame.to_sql(name, con, *, schema=None, if_exists='fail', index=True, index_label=None, chunksize=None, dtype=None, method=None)[源代码]#

将存储在 DataFrame 中的记录写入 SQL 数据库。

SQLAlchemy 支持的数据库 [1] 均受支持。表可以新建、追加或覆盖。

警告

pandas 库不会尝试清理通过 to_sql 调用提供的值。请参考底层数据库驱动的文档，查看其是否能正确防止注入，否则在执行 to_sql 调用中的任意命令时需注意安全风险。

参数:
namestr

SQL 表的名称。

conADBC connection, sqlalchemy.engine.(Engine or Connection) 或 sqlite3.Connection

ADBC 提供高性能 I/O，并支持原生类型（如果可用）。使用 SQLAlchemy 可以使用该库支持的任何数据库。为 sqlite3.Connection 对象提供遗留支持。用户负责 SQLAlchemy connectable 的引擎处置和连接关闭。详情请参见 此处。如果传递一个已处于事务中的 sqlalchemy.engine.Connection，则该事务不会被提交。如果传递一个 sqlite3.Connection，则无法回滚记录插入。

schemastr, optional

指定 schema（如果数据库支持此功能）。如果为 None，则使用默认 schema。

if_exists{‘fail’, ‘replace’, ‘append’, ‘delete_rows’}, default ‘fail’

如果表已存在时如何处理。

  • fail: 引发 ValueError。

  • replace: 在插入新值之前删除表。

  • append: 向现有表插入新值。

  • delete_rows: 如果表已存在，则删除所有记录并插入数据。

indexbool, 默认 True

将 DataFrame 索引写入为一列。使用 index_label 作为表中的列名。为该列创建表索引。

index_labelstr or sequence, default None

索引列的列标签。如果给定 None（默认值）且 index 为 True，则使用索引名称。如果 DataFrame 使用 MultiIndex，则应给出序列。

chunksize整数，可选

指定一次性写入数据库连接的行数。默认情况下，所有行将一次性写入。另请参见 method 关键字。

dtypedict or scalar, optional

指定列的数据类型。如果使用字典，键应为列名，值应为 SQLAlchemy 类型或 sqlite3 遗留模式的字符串。如果提供标量，则将其应用于所有列。

method{None, ‘multi’, callable}, optional

控制使用的 SQL 插入语句

  • None : 使用标准的 SQL INSERT 语句（每行一个）。

  • ‘multi’: 在单个 INSERT 语句中传递多个值。

  • 签名 (pd_table, conn, keys, data_iter) 的可调用对象。

有关详细信息和可调用示例实现，请参见 插入方法 部分。

返回:
None 或 int

to_sql 影响的行数。如果传递给 method 的可调用对象未返回整数行数，则返回 None。

返回的受影响行数是 sqlite3.Cursor 或 SQLAlchemy connectable 的 rowcount 属性的总和，该属性可能无法反映写入的实际行数，如 sqlite3 或 SQLAlchemy 中所述。

引发:
ValueError

当表已存在且 if_exists 为 ‘fail’（默认值）时。

另请参阅

read_sql

从表中读取 DataFrame。

注意

如果数据库支持，时区感知的 datetime 列将作为具有时区的 Timestamp 类型通过 SQLAlchemy 写入。否则，datetimes 将作为与原始时区本地化的时区不敏感的时间戳存储。

并非所有数据存储都支持 method="multi"。例如，Oracle 不支持多值插入。

参考文献

示例

创建一个内存中的 SQLite 数据库。

>>> from sqlalchemy import create_engine
>>> engine = create_engine('sqlite://', echo=False)

从头开始创建一个包含 3 行的表。

>>> df = pd.DataFrame({'name' : ['User 1', 'User 2', 'User 3']})
>>> df
     name
0  User 1
1  User 2
2  User 3

>>> df.to_sql(name='users', con=engine)
3
>>> from sqlalchemy import text
>>> with engine.connect() as conn:
...     conn.execute(text("SELECT * FROM users")).fetchall()
[(0, 'User 1'), (1, 'User 2'), (2, 'User 3')]

也可以将 sqlalchemy.engine.Connection 传递给 con

>>> with engine.begin() as connection:
...     df1 = pd.DataFrame({'name' : ['User 4', 'User 5']})
...     df1.to_sql(name='users', con=connection, if_exists='append')
2

这允许支持需要在整个操作中使用相同 DBAPI 连接的操作。

>>> df2 = pd.DataFrame({'name' : ['User 6', 'User 7']})
>>> df2.to_sql(name='users', con=engine, if_exists='append')
2
>>> with engine.connect() as conn:
...     conn.execute(text("SELECT * FROM users")).fetchall()
[(0, 'User 1'), (1, 'User 2'), (2, 'User 3'),
 (0, 'User 4'), (1, 'User 5'), (0, 'User 6'),
 (1, 'User 7')]

用 df2 覆盖表。

>>> df2.to_sql(name='users', con=engine, if_exists='replace',
...            index_label='id')
2
>>> with engine.connect() as conn:
...     conn.execute(text("SELECT * FROM users")).fetchall()
[(0, 'User 6'), (1, 'User 7')]

删除所有行，然后用 df3 插入新记录

>>> df3 = pd.DataFrame({"name": ['User 8', 'User 9']})
>>> df3.to_sql(name='users', con=engine, if_exists='delete_rows',
...            index_label='id')
2
>>> with engine.connect() as conn:
...     conn.execute(text("SELECT * FROM users")).fetchall()
[(0, 'User 8'), (1, 'User 9')]

在 PostgreSQL 数据库中，使用 method 定义一个可调用插入方法，如果存在主键冲突则不执行任何操作。

>>> from sqlalchemy.dialects.postgresql import insert
>>> def insert_on_conflict_nothing(table, conn, keys, data_iter):
...     # "a" is the primary key in "conflict_table"
...     data = [dict(zip(keys, row)) for row in data_iter]
...     stmt = insert(table.table).values(data).on_conflict_do_nothing(index_elements=["a"])
...     result = conn.execute(stmt)
...     return result.rowcount
>>> df_conflict.to_sql(name="conflict_table", con=conn, if_exists="append",  # noqa: F821
...                    method=insert_on_conflict_nothing)
0

对于 MySQL，提供一个可调用函数，在主键冲突时更新列 b 和 c。

>>> from sqlalchemy.dialects.mysql import insert   # noqa: F811
>>> def insert_on_conflict_update(table, conn, keys, data_iter):
...     # update columns "b" and "c" on primary key conflict
...     data = [dict(zip(keys, row)) for row in data_iter]
...     stmt = (
...         insert(table.table)
...         .values(data)
...     )
...     stmt = stmt.on_duplicate_key_update(b=stmt.inserted.b, c=stmt.inserted.c)
...     result = conn.execute(stmt)
...     return result.rowcount
>>> df_conflict.to_sql(name="conflict_table", con=conn, if_exists="append",  # noqa: F821
...                    method=insert_on_conflict_update)
2

指定 dtype（对于带有缺失值的整数特别有用）。请注意，虽然 pandas 被迫将数据存储为浮点数，但数据库支持可为空的整数。在通过 Python 获取数据时，我们得到的是整数标量。

>>> df = pd.DataFrame({"A": [1, None, 2]})
>>> df
     A
0  1.0
1  NaN
2  2.0

>>> from sqlalchemy.types import Integer
>>> df.to_sql(name='integers', con=engine, index=False,
...           dtype={"A": Integer()})
3

>>> with engine.connect() as conn:
...     conn.execute(text("SELECT * FROM integers")).fetchall()
[(1,), (None,), (2,)]

2.2.0 版本已添加: pandas 现在支持通过 ADBC 驱动程序进行写入

>>> df = pd.DataFrame({'name' : ['User 10', 'User 11', 'User 12']})
>>> df
      name
0  User 10
1  User 11
2  User 12

>>> from adbc_driver_sqlite import dbapi
>>> with dbapi.connect("sqlite://") as conn:
...     df.to_sql(name="users", con=conn)
3
本页内容