(
source_connection: Connection,
target_connection: Connection,
plan: TablePlan,
batch_size: int,
)
| 135 | |
| 136 | |
| 137 | def copy_table( |
| 138 | source_connection: Connection, |
| 139 | target_connection: Connection, |
| 140 | plan: TablePlan, |
| 141 | batch_size: int, |
| 142 | ) -> TableCopyResult: |
| 143 | rows = fetch_source_rows(source_connection, plan) |
| 144 | inserted_rows = 0 |
| 145 | if rows: |
| 146 | insert_stmt = plan.target_table.insert() |
| 147 | for chunk in chunked_rows(rows, batch_size): |
| 148 | target_connection.execute(insert_stmt, chunk) |
| 149 | inserted_rows += len(chunk) |
| 150 | return TableCopyResult( |
| 151 | name=plan.name, |
| 152 | source_rows=len(rows), |
| 153 | inserted_rows=inserted_rows, |
| 154 | columns=plan.columns, |
| 155 | ) |
| 156 | |
| 157 | |
| 158 | def reset_postgres_sequences(connection: Connection, plan: Sequence[TablePlan]) -> None: |
no test coverage detected