(self, transaction: Transaction, logging_exe: Any | None = None)
| 788 | ) |
| 789 | |
| 790 | def run(self, transaction: Transaction, logging_exe: Any | None = None) -> None: |
| 791 | self.logging_exe = logging_exe |
| 792 | with self.pg_conn.cursor() as cur: |
| 793 | for row_list in transaction.row_lists: |
| 794 | for row in row_list.rows: |
| 795 | if row.operation == Operation.INSERT: |
| 796 | values_str = ", ".join( |
| 797 | str(formatted_value(value)) for value in row.values |
| 798 | ) |
| 799 | self.execute( |
| 800 | cur, |
| 801 | f"""INSERT INTO {identifier(self.table)} |
| 802 | VALUES ({values_str}) |
| 803 | """, |
| 804 | ) |
| 805 | elif row.operation == Operation.UPSERT: |
| 806 | values_str = ", ".join( |
| 807 | str(formatted_value(value)) for value in row.values |
| 808 | ) |
| 809 | keys_str = ", ".join( |
| 810 | identifier(field.name) |
| 811 | for field in row.fields |
| 812 | if field.is_key |
| 813 | ) |
| 814 | update_str = ", ".join( |
| 815 | f"{identifier(field.name)} = EXCLUDED.{identifier(field.name)}" |
| 816 | for field in row.fields |
| 817 | ) |
| 818 | self.execute( |
| 819 | cur, |
| 820 | f"""INSERT INTO {identifier(self.table)} |
| 821 | VALUES ({values_str}) |
| 822 | ON CONFLICT ({keys_str}) |
| 823 | DO UPDATE SET {update_str} |
| 824 | """, |
| 825 | ) |
| 826 | elif row.operation == Operation.DELETE: |
| 827 | cond_str = " AND ".join( |
| 828 | f"{identifier(field.name)} = {formatted_value(value)}" |
| 829 | for field, value in zip(row.fields, row.values) |
| 830 | if field.is_key |
| 831 | ) |
| 832 | self.execute( |
| 833 | cur, |
| 834 | f"""DELETE FROM {identifier(self.table)} |
| 835 | WHERE {cond_str} |
| 836 | """, |
| 837 | ) |
| 838 | else: |
| 839 | raise ValueError(f"Unexpected operation {row.operation}") |
| 840 | self.pg_conn.commit() |
| 841 | |
| 842 | |
| 843 | class KafkaRoundtripExecutor(Executor): |
nothing calls this directly
no test coverage detected