(
cfg,
table,
returning,
*,
set_lambda=None,
sort_by_parameter_order=False,
index_elements=None,
)
| 100 | |
| 101 | @upsert.for_db("mariadb") |
| 102 | def _upsert( |
| 103 | cfg, |
| 104 | table, |
| 105 | returning, |
| 106 | *, |
| 107 | set_lambda=None, |
| 108 | sort_by_parameter_order=False, |
| 109 | index_elements=None, |
| 110 | ): |
| 111 | from sqlalchemy.dialects.mysql import insert |
| 112 | |
| 113 | stmt = insert(table) |
| 114 | |
| 115 | if set_lambda: |
| 116 | stmt = stmt.on_duplicate_key_update(**set_lambda(stmt.inserted)) |
| 117 | else: |
| 118 | pk1 = table.primary_key.c[0] |
| 119 | stmt = stmt.on_duplicate_key_update({pk1.key: pk1}) |
| 120 | |
| 121 | stmt = stmt.returning( |
| 122 | *returning, sort_by_parameter_order=sort_by_parameter_order |
| 123 | ) |
| 124 | return stmt |
| 125 | |
| 126 | |
| 127 | @delete_from_all_tables.for_db("mysql", "mariadb") |
nothing calls this directly
no test coverage detected