我想使用SQLAlchemy删除表。
由于我要一遍又一遍地进行测试,因此我想删除该表,my_users
以便每次都可以从头开始。
到目前为止,我正在使用SQLAlchemy通过engine.execute()方法执行原始SQL :
sql = text('DROP TABLE IF EXISTS my_users;')
result = engine.execute(sql)
但是,我想知道是否有一些标准方法可以做到这一点。我能找到的唯一一个是drop_all()
,但是它删除了所有结构,而不仅删除了一个特定的表:
Base.metadata.drop_all(engine) # all tables are deleted
例如,给出这个非常基本的例子。它由一个带有单个表的SQLite基础结构组成my_users
,我在其中添加了一些内容。
from sqlalchemy import create_engine, Column, Integer, String, text
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
engine = create_engine('sqlite://', echo=False)
Base = declarative_base()
class User(Base):
__tablename__ = "my_users"
id = Column(Integer, primary_key=True)
name = Column(String)
def __init__(self, name):
self.name = name
# Create all the tables in the database which are
# defined by Base's subclasses such as User
Base.metadata.create_all(engine)
# Construct a sessionmaker factory object
session = sessionmaker()
# Bind the sessionmaker to engine
session.configure(bind=engine)
# Generate a session to work with
s = session()
# Add some content
s.add(User('myname'))
s.commit()
# Fetch the data
print(s.query(User).filter(User.name == 'myname').one().name)
对于这种特定情况,它drop_all()
可以工作,但是从我开始拥有多个表并保留其他表的那一刻起,它就不方便了。
只需drop()
对表对象进行调用。从文档:
使用给定的Connectable进行连接,为此表发出DROP语句。
在您的情况下,应为:
User.__table__.drop()
如果出现类似以下的异常:
sqlalchemy.exc.UnboundExecutionError: Table object 'my_users' is not bound to an Engine or Connection. Execution can not proceed without a database to execute against
您需要传递引擎:
User.__table__.drop(engine)
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句