| 2854 | cursor.close() |
| 2855 | |
| 2856 | def get_isolation_level( |
| 2857 | self, dbapi_connection: DBAPIConnection |
| 2858 | ) -> IsolationLevel: |
| 2859 | cursor = dbapi_connection.cursor() |
| 2860 | if self._is_mysql and self.server_version_info >= (5, 7, 20): |
| 2861 | cursor.execute("SELECT @@transaction_isolation") |
| 2862 | else: |
| 2863 | cursor.execute("SELECT @@tx_isolation") |
| 2864 | row = cursor.fetchone() |
| 2865 | if row is None: |
| 2866 | util.warn( |
| 2867 | "Could not retrieve transaction isolation level for MySQL " |
| 2868 | "connection." |
| 2869 | ) |
| 2870 | raise NotImplementedError() |
| 2871 | val = row[0] |
| 2872 | cursor.close() |
| 2873 | if isinstance(val, bytes): |
| 2874 | val = val.decode() |
| 2875 | return val.upper().replace("-", " ") # type: ignore[no-any-return] |
| 2876 | |
| 2877 | def _get_server_version_info( |
| 2878 | self, connection: Connection |