合并两个不同数据库中的表- sqlite3/Python

azpvetkf  于 2022-11-24  发布在  SQLite
关注(0)|答案(2)|浏览(254)

我有两个不同的SQLite数据库XXX和YYY。XXX包含表A,YYY包含表B。A和B有相同的结构(列)。如何在Python中追加B的行- SQLite API。追加后A包含A的行和B的行。

l7wslrjt

l7wslrjt1#

首先使用sqlite3.connect获得到数据库的连接,然后创建一个游标以便执行sql。一旦有了游标,就可以执行任意sql命令。
示例:

import sqlite3

# Get connections to the databases
db_a = sqlite3.connect('database_a.db')
db_b = sqlite3.connect('database_b.db')

# Get the contents of a table
b_cursor = db_b.cursor()
b_cursor.execute('SELECT * FROM mytable')
output = b_cursor.fetchall()   # Returns the results as a list.

# Insert those contents into another table.
a_cursor = db_a.cursor()
for row in output:
    a_cursor.execute('INSERT INTO myothertable VALUES (?, ?, ...etc..., ?, ?)', row)

# Cleanup
db_a.commit()
a_cursor.close()
b_cursor.close()

注意:我还没有真正测试过这个,所以它可能有一些bug,但我认为基本的想法是合理的。

lndjwyie

lndjwyie2#

这是一个通用函数,应针对您的特定环境进行定制。为此,您可以使用静态SQL参数(而不是PRAGMA table_info)来构建“动态确定SQL表达式要求”部分。这应该可以提高性能。

import sqlite3

def merge_tables(cursor_new: sqlite3.Cursor, cursor_old: sqlite3.Cursor, table_name: str, del_old_table: bool = False) -> None:
    '''
    This function merges the content of a specific table from an old cursor into a new cursor. 
    
    :param cursor_new: [sqlite3.Cursor] the primary cursor
    :param cursor_old: [sqlite3.Cursor] the secondary cursor
    :param table_name: [str] the name of the table
    :return: None
    '''

    # dynamically determine SQL expression requirements
    column_names = cursor_new.execute(f"PRAGMA table_info({table_name})").fetchall()
    column_names = tuple([x[1] for x in column_names][1:])  # remove the primary keyword
    values_placeholders = ', '.join(['?' for x in column_names])  # format appropriately
    
    # SQL select columns from table
    data = cursor_old.execute(f"SELECT {', '.join(column_names)} FROM {table_name}").fetchall()

    # insert the data into the primary cursor
    cursor_new.executemany(f"INSERT INTO {table_name} {column_names} VALUES ({values_placeholders})", data)
    if (cursor_new.connection.commit() == None):
        # With Ephemeral RAM connections & testing, deleting the table may be ill-advised
        if del_old_table:
            cursor_old.execute(f"DELETE FROM {table_name}")  # cursor_old.execute(f'DROP TABLE {table_name}')
            cursor_old.connection.commit()
        print(f"Table {table_name} merged from {cursor_old.connection} to {cursor_new.connection}") # Consider logging.info()
    return None

相关问题