(
self,
connection: Connection,
sequence_name: str,
schema: Optional[str] = None,
**kw: Any,
)
| 3099 | |
| 3100 | @reflection.cache |
| 3101 | def has_sequence( |
| 3102 | self, |
| 3103 | connection: Connection, |
| 3104 | sequence_name: str, |
| 3105 | schema: Optional[str] = None, |
| 3106 | **kw: Any, |
| 3107 | ) -> bool: |
| 3108 | if not self.supports_sequences: |
| 3109 | self._sequences_not_supported() |
| 3110 | if not schema: |
| 3111 | schema = self.default_schema_name |
| 3112 | # MariaDB implements sequences as a special type of table |
| 3113 | # |
| 3114 | cursor = connection.execute( |
| 3115 | sql.text( |
| 3116 | "SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES " |
| 3117 | "WHERE TABLE_TYPE='SEQUENCE' and TABLE_NAME=:name AND " |
| 3118 | "TABLE_SCHEMA=:schema_name" |
| 3119 | ), |
| 3120 | dict( |
| 3121 | name=str(sequence_name), |
| 3122 | schema_name=str(schema), |
| 3123 | ), |
| 3124 | ) |
| 3125 | return cursor.first() is not None |
| 3126 | |
| 3127 | def _sequences_not_supported(self) -> NoReturn: |
| 3128 | raise NotImplementedError( |
nothing calls this directly
no test coverage detected