(connection: Connection, table_names: Iterable[str])
| 110 | |
| 111 | |
| 112 | def truncate_target_tables(connection: Connection, table_names: Iterable[str]) -> None: |
| 113 | quoted_names = [quote_identifier(connection, name) for name in table_names] |
| 114 | if not quoted_names: |
| 115 | return |
| 116 | connection.execute(text(f"TRUNCATE TABLE {', '.join(quoted_names)} RESTART IDENTITY CASCADE")) |
| 117 | |
| 118 | |
| 119 | def ensure_target_empty(connection: Connection, plan: Sequence[TablePlan]) -> None: |
no test coverage detected