Cómo ejecutar SQL sin procesar en SQLAlchemy

En este artículo, veremos cómo escribir una consulta SQL convencional en SQLAlchemy usando text() contra una base de datos PostgreSQL en python.

Creación de tabla para demostración.

Importe las funciones necesarias del paquete SQLAlchemy. Establezca una conexión con la base de datos PostgreSQL usando la función create_engine() como se muestra a continuación, cree una tabla llamada libros con las columnas book_id y book_price.

Inserte un registro en las tablas usando la función insert() y valores() como se muestra.

# import necessary packages
import sqlalchemy
from sqlalchemy import create_engine, MetaData,
Table, Column, Numeric, Integer, VARCHAR
from sqlalchemy.engine import result
  
# establish connections
engine = create_engine(
    "database+dialect://username:password@hostname:port/databasename")
  
# initialize the Metadata Object
meta = MetaData(bind=engine)
MetaData.reflect(meta)
  
# create a table schema
books = Table(
    'books', meta,
    Column('book_id', Integer, primary_key=True),
    Column('book_price', Numeric),
    Column('genre', VARCHAR),
    Column('book_name', VARCHAR)
)
  
meta.create_all(engine)
  
# insert records into the table
statement1 = books.insert().values(book_id=1, book_price=12.2,
                                   genre='fiction', 
                                   book_name='Old age')
statement2 = books.insert().values(book_id=2, book_price=13.2,
                                   genre='non-fiction',
                                   book_name='Saturn rings')
statement3 = books.insert().values(book_id=3, book_price=121.6,
                                   genre='fiction', 
                                   book_name='Supernova')
statement4 = books.insert().values(book_id=4, book_price=100,
                                   genre='non-fiction', 
                                   book_name='History of the world')
statement5 = books.insert().values(book_id=5, book_price=1112.2,
                                   genre='fiction', book_name='Sun city')
  
# execute the insert records statement
engine.execute(statement1)
engine.execute(statement2)
engine.execute(statement3)
engine.execute(statement4)
engine.execute(statement5)

Producción:

tabla de muestra

Ejemplo 1: implementar una consulta para extraer filas completas en SQLAlchemy

SQLAlchemy proporciona una función llamada text(). Podemos escribir cualquier consulta SQL convencional dentro de la función de texto encerrada por “”. Ahora, al pasar esta consulta SQL para ejecutar la función, el objeto del motor que creamos mientras conectamos la base de datos interpretará esto y convertirá esta consulta a un formato compatible con SQLAlchemy y dará como resultado el resultado.

from sqlalchemy import text
text("YOUR SQL QUERY")

Pase la consulta SQL a la función execute() y obtenga todos los resultados usando la función fetchall(). Use un bucle for para iterar a través de los resultados. La consulta SQLAlchemy que se muestra en el siguiente código selecciona todas las filas donde el precio del libro es mayor que Rs. 50

from sqlalchemy import text
  
# write the SQL query inside the text() block
sql = text('SELECT * from BOOKS WHERE BOOKS.book_price > 50')
results = engine.execute(sql)
  
# View the records
for record in results:
    print("\n", record)

Producción:

El resultado de una consulta SQL sin formato en SQLALchemy

Ejemplo 2: Insertar un registro utilizando una consulta SQL sin procesar en SQLAlchemy

La consulta SQLAlchemy que se muestra en el siguiente código selecciona todas las filas donde el precio del libro es mayor que Rs. 50

from sqlalchemy import text
  
# define a tuple of dictionary of 
# values to be inserted
data = ( { "book_id": 6,
          "book_price": 400,
          "genre": "fiction",
          "book_name": "yoga is science" },
          
         { "book_id": 7, 
          "book_price": 800, 
          "genre": "non-fiction",
          "book_name": "alchemy tutorials" },
)
  
# write the insert statement
statement = text("""INSERT INTO BOOKS (book_id, book_price,
genre, book_name) VALUES(:book_id, :book_price,
:genre, :book_name)""")
  
#insert the data one after other using execute
# statement by unpacking dictionary  elements
for line in data:
    engine.execute(statement, **line)
  
# write the SQL query to check
# whether the records are inserted
sql = text("SELECT * FROM BOOKS ")
  
results = engine.execute(sql)
  
# View the records
for record in results:
    print("\n", record)

Producción:

La salida de la consulta de inserción