如何使用SQLAlchemy实现存在则删除表（DROP TABLE IF EXISTS）？

问题场景 #

先列出sample数据库中的所有表：

from sqlalchemy import (MetaData, Table)
from sqlalchemy import create_engine, inspect
from sqlalchemy.ext.declarative import declarative_base

db_pass = 'xxxxxx'
db_ip='127.0.0.1'
engine = create_engine('postgresql://postgres:{}@{}/sample'.format(db_pass,db_ip))
inspector = inspect(engine)
print(inspector.get_table_names())
# 输出: ['pre', 'sample']

数据库中有pre和sample两张表，尝试删除pre表时执行了以下代码：

Base = declarative_base()
metadata = MetaData(engine)
if inspector.has_table('pre'):
    Base.metadata.drop_all(engine, ['pre'], checkfirst=True)

执行后出现错误：

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/sql/schema.py", line 4959, in drop_all
    bind._run_ddl_visitor(
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 3228, in _run_ddl_visitor
    conn._run_ddl_visitor(visitorcallable, element, **kwargs)
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/engine/base.py", line 2211, in _run_ddl_visitor
    visitorcallable(self.dialect, self, **kwargs).traverse_single(element)
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/sql/visitors.py", line 524, in traverse_single
    return meth(obj, **kw)
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py", line 959, in visit_metadata
    unsorted_tables = [t for t in tables if self._can_drop_table(t)]
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py", line 959, in <listcomp>
    unsorted_tables = [t for t in tables if self._can_drop_table(t)]
  File "/home/debian/.local/lib/python3.9/site-packages/sqlalchemy/sql/ddl.py", line 1047, in _can_drop_table
    self.dialect.validate_identifier(table.name)
AttributeError: 'str' object has no attribute 'name'

错误原因 #

drop_all方法的tables参数要求传入SQLAlchemy的Table对象，而非字符串。直接传['pre']会导致代码尝试访问字符串的name属性，从而抛出AttributeError。

解决方案 #

方案一：反射表对象后删除 #

通过MetaData反射出目标表，再传入drop_all：

from sqlalchemy import MetaData, create_engine, inspect

db_pass = 'xxxxxx'
db_ip='127.0.0.1'
engine = create_engine('postgresql://postgres:{}@{}/sample'.format(db_pass,db_ip))
inspector = inspect(engine)

metadata = MetaData()
if inspector.has_table('pre'):
    # 反射pre表
    pre_table = Table('pre', metadata, autoload_with=engine)
    # 删除该表
    metadata.drop_all(engine, tables=[pre_table])

方案二：直接执行原生SQL #

如果不需要ORM对象操作，可直接执行原生DROP TABLE IF EXISTS语句：

from sqlalchemy import create_engine

db_pass = 'xxxxxx'
db_ip='127.0.0.1'
engine = create_engine('postgresql://postgres:{}@{}/sample'.format(db_pass,db_ip))

with engine.connect() as conn:
    # 执行原生SQL
    conn.execute("DROP TABLE IF EXISTS pre")
    conn.commit()

方案三：检查表存在后直接删除表对象 #

结合inspector确认表存在，再通过Table对象执行删除：

from sqlalchemy import Table, MetaData, create_engine, inspect

db_pass = 'xxxxxx'
db_ip='127.0.0.1'
engine = create_engine('postgresql://postgres:{}@{}/sample'.format(db_pass,db_ip))
inspector = inspect(engine)

if inspector.has_table('pre'):
    metadata = MetaData()
    pre_table = Table('pre', metadata, autoload_with=engine)
    pre_table.drop(engine, checkfirst=True)

内容的提问来源于stack exchange，提问作者showkey