python - query - ¿Cómo borrar una tabla en SQLAlchemy?
sqlalchemy select (4)
Quiero borrar una tabla usando SQLAlchemy.
Ya que estoy probando una y otra vez, quiero eliminar la tabla my_users
para poder comenzar desde cero cada vez.
Hasta ahora estoy usando SQLAlchemy para ejecutar SQL en bruto a través del método engine.execute() :
sql = text(''DROP TABLE IF EXISTS my_users;'')
result = engine.execute(sql)
Sin embargo, me pregunto si hay alguna forma estándar de hacerlo. El único que pude encontrar es drop_all()
, pero elimina toda la estructura, no solo una tabla específica:
Base.metadata.drop_all(engine) # all tables are deleted
Por ejemplo, dado este ejemplo muy básico. Consiste en una infraestructura SQLite con una sola tabla my_users
en la que agrego contenido.
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)
Para este caso específico, drop_all()
funcionaría, pero no será conveniente desde el momento en que comience a tener más de una mesa y quiero conservar las otras.
A continuación se muestra un código de ejemplo que puede ejecutar en iPython para probar la creación y eliminación de una tabla en Postgres
from sqlalchemy import * # imports all needed modules from sqlalchemy
engine = create_engine(''postgresql://python:[email protected]/production'') # connection properties stored
metadata = MetaData() # stores the ''production'' database''s metadata
users = Table(''users'', metadata,
Column(''user_id'', Integer),
Column(''first_name'', String(150)),
Column(''last_name'', String(150)),
Column(''email'', String(255)),
schema=''python'') # defines the ''users'' table structure in the ''python'' schema of our connection to the ''production'' db
users.create(engine) # creates the users table
users.drop(engine) # drops the users table
También puede obtener una vista previa de mi artículo en Wordpress con este mismo ejemplo y capturas de pantalla: oscarvalles.wordpress.com (búsqueda de Alquimia de SQL).
Alternativa a llamar cls.__table__.drop(your_engine)
, puede intentar esto:
Base.metadata.drop_all(bind=your_engine, tables=[User.__table__])
Este método, así como el método create_all()
, aceptan tables
argumentos opcionales, que toman un iterador de instancias de sqlalchemy.sql.schema.Table
.
Puede controlar qué tablas se crearán o eliminarán de esta manera.
Para el caso especial cuando no tiene acceso a la clase de tabla y solo necesita eliminar la tabla por nombre de tabla, puede usar este código
import logging
from sqlalchemy import MetaData
from sqlalchemy import create_engine
from sqlalchemy.engine.url import URL
from sqlalchemy.ext.declarative import declarative_base
DATABASE = {
''drivername'': ''sqlite'',
# ''host'': ''localhost'',
# ''port'': ''5432'',
# ''username'': ''YOUR_USERNAME'',
# ''password'': ''YOUR_PASSWORD'',
''database'': ''/path/to/your_db.sqlite''
}
def drop_table(table_name):
engine = create_engine(URL(**DATABASE))
base = declarative_base()
metadata = MetaData(engine, reflect=True)
table = metadata.tables.get(table_name)
if table is not None:
logging.info(f''Deleting {table_name} table'')
base.metadata.drop_all(engine, [table], checkfirst=True)
drop_table(''users'')
Simplemente llame a drop()
contra el objeto de la tabla. De los documentos :
Emita una declaración DROP para esta Tabla, utilizando el Conectable dado para conectividad.
En tu caso debería ser:
User.__table__.drop()
Si obtiene una excepción como:
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
Necesitas pasar el motor:
User.__table__.drop(engine)