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