(self, style, tables, *, reset_sequences=False, allow_cascade=False)
| 204 | return lru_cache(maxsize=512)(self.__references_graph) |
| 205 | |
| 206 | def sql_flush(self, style, tables, *, reset_sequences=False, allow_cascade=False): |
| 207 | if tables and allow_cascade: |
| 208 | # Simulate TRUNCATE CASCADE by recursively collecting the tables |
| 209 | # referencing the tables to be flushed. |
| 210 | tables = set( |
| 211 | chain.from_iterable(self._references_graph(table) for table in tables) |
| 212 | ) |
| 213 | sql = [ |
| 214 | "%s %s %s;" |
| 215 | % ( |
| 216 | style.SQL_KEYWORD("DELETE"), |
| 217 | style.SQL_KEYWORD("FROM"), |
| 218 | style.SQL_FIELD(self.quote_name(table)), |
| 219 | ) |
| 220 | for table in tables |
| 221 | ] |
| 222 | if reset_sequences: |
| 223 | sequences = [{"table": table} for table in tables] |
| 224 | sql.extend(self.sequence_reset_by_name_sql(style, sequences)) |
| 225 | return sql |
| 226 | |
| 227 | def sequence_reset_by_name_sql(self, style, sequences): |
| 228 | if not sequences: |
nothing calls this directly
no test coverage detected