(self, user_address_fixture, flush_user_first)
| 1865 | @testing.requires.delete_returning |
| 1866 | @testing.combinations(True, False, argnames="flush_user_first") |
| 1867 | def test_bulk_delete(self, user_address_fixture, flush_user_first): |
| 1868 | User, Address = user_address_fixture( |
| 1869 | addresses_args={"backref": "user"} |
| 1870 | ) |
| 1871 | sess = fixture_session() |
| 1872 | |
| 1873 | u1 = User( |
| 1874 | name="x", |
| 1875 | addresses=[ |
| 1876 | Address(email_address="e1"), |
| 1877 | Address(email_address="e2"), |
| 1878 | Address(email_address="e3"), |
| 1879 | ], |
| 1880 | ) |
| 1881 | sess.add(u1) |
| 1882 | |
| 1883 | # ha ha! u1 is not persistent yet. autoflush won't happen |
| 1884 | # until sess.scalars() actually runs. statement has to be |
| 1885 | # created with a pending parameter, not actual parameter |
| 1886 | assert inspect(u1).pending |
| 1887 | |
| 1888 | if flush_user_first: |
| 1889 | sess.flush() |
| 1890 | |
| 1891 | with self.sql_execution_asserter() as asserter: |
| 1892 | addresses = sess.scalars( |
| 1893 | u1.addresses.delete() |
| 1894 | .where(Address.email_address == "e2") |
| 1895 | .returning(Address), |
| 1896 | ).all() |
| 1897 | |
| 1898 | eq_( |
| 1899 | addresses, |
| 1900 | [ |
| 1901 | Address(email_address="e2"), |
| 1902 | ], |
| 1903 | ) |
| 1904 | |
| 1905 | uid = u1.id |
| 1906 | |
| 1907 | asserter.assert_( |
| 1908 | Conditional( |
| 1909 | not flush_user_first, |
| 1910 | [ |
| 1911 | CompiledSQL( |
| 1912 | "INSERT INTO users (name) VALUES (:name)", |
| 1913 | [{"name": "x"}], |
| 1914 | ), |
| 1915 | Conditional( |
| 1916 | testing.requires.insert_executemany_returning.enabled, |
| 1917 | [ |
| 1918 | CompiledSQL( |
| 1919 | "INSERT INTO addresses " |
| 1920 | "(user_id, email_address) " |
| 1921 | "VALUES (:user_id, :email_address) " |
| 1922 | "RETURNING addresses.id", |
| 1923 | [ |
| 1924 | {"user_id": uid, "email_address": "e1"}, |
nothing calls this directly
no test coverage detected