(self, returning_fields=None)
| 1922 | ] |
| 1923 | |
| 1924 | def execute_sql(self, returning_fields=None): |
| 1925 | assert not ( |
| 1926 | returning_fields |
| 1927 | and len(self.query.objs) != 1 |
| 1928 | and not self.connection.features.can_return_rows_from_bulk_insert |
| 1929 | ) |
| 1930 | opts = self.query.get_meta() |
| 1931 | self.returning_fields = returning_fields |
| 1932 | cols = [] |
| 1933 | with self.connection.cursor() as cursor: |
| 1934 | for sql, params in self.as_sql(): |
| 1935 | cursor.execute(sql, params) |
| 1936 | if not self.returning_fields: |
| 1937 | return [] |
| 1938 | obj_len = len(self.query.objs) |
| 1939 | if ( |
| 1940 | self.connection.features.can_return_rows_from_bulk_insert |
| 1941 | and obj_len > 1 |
| 1942 | ) or ( |
| 1943 | self.connection.features.can_return_columns_from_insert and obj_len == 1 |
| 1944 | ): |
| 1945 | rows = self.connection.ops.fetch_returned_rows( |
| 1946 | cursor, self.returning_params |
| 1947 | ) |
| 1948 | cols = [field.get_col(opts.db_table) for field in self.returning_fields] |
| 1949 | elif returning_fields and isinstance( |
| 1950 | returning_field := returning_fields[0], AutoField |
| 1951 | ): |
| 1952 | cols = [returning_field.get_col(opts.db_table)] |
| 1953 | rows = [ |
| 1954 | ( |
| 1955 | self.connection.ops.last_insert_id( |
| 1956 | cursor, |
| 1957 | opts.db_table, |
| 1958 | returning_field.column, |
| 1959 | ), |
| 1960 | ) |
| 1961 | ] |
| 1962 | else: |
| 1963 | # Backend doesn't support returning fields and no auto-field |
| 1964 | # that can be retrieved from `last_insert_id` was specified. |
| 1965 | return [] |
| 1966 | converters = self.get_converters(cols) |
| 1967 | if converters: |
| 1968 | rows = self.apply_converters(rows, converters) |
| 1969 | return list(rows) |
| 1970 | |
| 1971 | |
| 1972 | class SQLDeleteCompiler(SQLCompiler): |
nothing calls this directly
no test coverage detected