I want to delete a table using SQLAlchemy.
Since I am testing over and over again, I want to delete the table my_users
so that I can start from scratch every single time.
So far I am using SQLAlchemy to execute raw SQL through the engine.execute() method:
sql = text('DROP TABLE IF EXISTS my_users;')
result = engine.execute(sql)
However, I wonder if there is some standard way to do so. The only one I could find is drop_all()
, but it deletes all the structure, not only one specific table:
Base.metadata.drop_all(engine) # all tables are deleted
For example, given this very basic example. It consists on a SQLite infrastructure with a single table my_users
in which I add some content.
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)
For this specific case, drop_all()
would work, but it won't be convenient from the moment I start having more than one table and I want to keep the other ones.
Just call drop()
against the table object.
From the docs:
Issue a DROP statement for this Table, using the given Connectable for connectivity.
In your case it should be:
User.__table__.drop()
If you get an exception like:
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
You need to pass the engine:
User.__table__.drop(engine)