MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / _get_table_sql

Method _get_table_sql

lib/sqlalchemy/dialects/sqlite/base.py:3013–3039  ·  view source on GitHub ↗
(self, connection, table_name, schema=None, **kw)

Source from the content-addressed store, hash-verified

3011
3012 @reflection.cache
3013 def _get_table_sql(self, connection, table_name, schema=None, **kw):
3014 if schema:
3015 schema_expr = "%s." % (
3016 self.identifier_preparer.quote_identifier(schema)
3017 )
3018 else:
3019 schema_expr = ""
3020 try:
3021 s = (
3022 "SELECT sql FROM "
3023 " (SELECT * FROM %(schema)ssqlite_master UNION ALL "
3024 " SELECT * FROM %(schema)ssqlite_temp_master) "
3025 "WHERE name = ? "
3026 "AND type in ('table', 'view')" % {"schema": schema_expr}
3027 )
3028 rs = connection.exec_driver_sql(s, (table_name,))
3029 except exc.DBAPIError:
3030 s = (
3031 "SELECT sql FROM %(schema)ssqlite_master "
3032 "WHERE name = ? "
3033 "AND type in ('table', 'view')" % {"schema": schema_expr}
3034 )
3035 rs = connection.exec_driver_sql(s, (table_name,))
3036 value = rs.scalar()
3037 if value is None and not self._is_sys_table(table_name):
3038 raise exc.NoSuchTableError(f"{schema_expr}{table_name}")
3039 return value
3040
3041 def _get_table_pragma(self, connection, pragma, table_name, schema=None):
3042 quote = self.identifier_preparer.quote_identifier

Callers 5

get_columnsMethod · 0.95
get_pk_constraintMethod · 0.95
get_foreign_keysMethod · 0.95
get_check_constraintsMethod · 0.95

Calls 4

_is_sys_tableMethod · 0.95
quote_identifierMethod · 0.80
exec_driver_sqlMethod · 0.45
scalarMethod · 0.45

Tested by

no test coverage detected