如何在 SQLAlchemy 中执行原始 SQL

在本文中，我们将看到如何在 SQLAlchemy 中使用 text() 针对Python中的 PostgreSQL 数据库编写常规 SQL 查询。

为演示创建表

从 SQLAlchemy 包中导入必要的函数。使用 create_engine()函数与 PostgreSQL 数据库建立连接，如下所示，创建一个名为 books 的表，其中包含 book_id 和 book_price 列。

如图所示，使用 insert() 和 values()函数将记录插入表中。

Python3

# 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)

输出：

样品表

示例 1：在 SQLAlchemy 中实现一个查询以提取完整的行

SQLAlchemy 提供了一个名为 text() 的函数。我们可以在用“”括起来的文本函数中编写任何常规的 SQL 查询。现在，传递这个 SQL 查询来执行我们在连接数据库时创建的引擎对象的函数将解释这个并将这个查询转换为 SQLAlchemy 兼容格式并产生结果。

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

将 SQL 查询传递给 execute()函数并使用 fetchall()函数获取所有结果。使用 for 循环遍历结果。下面代码中显示的 SQLAlchemy 查询选择了书价大于 Rs 的所有行。 50.

Python3

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)

输出：

SQLALchemy 中原始 SQL 查询的输出

示例 2：在 SQLAlchemy 中使用原始 SQL 查询插入记录

下面代码中显示的 SQLAlchemy 查询选择了书价大于 Rs 的所有行。 50.

Python3

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)

输出：

插入查询的输出