适用于 Python 的亚马逊QLDB驱动程序 — 快速入门教程

重要

终止支持通知：现有客户将能够使用亚马逊，QLDB直到 2025 年 7 月 31 日终止支持。有关更多详细信息，请参阅将亚马逊QLDB账本迁移到亚马逊 Aurora Postgr SQL e。

在本教程中，您将学习如何使用最新版本的 Amazon Python QLDB 驱动程序设置简单的应用程序。本指南包括安装驱动程序的步骤以及基本创建、读取、更新和删除 (CRUD) 操作的简短代码示例。有关在完整示例应用程序中演示这些操作的更深入的示例，请参阅 Python 教程。

先决条件

在开始之前，请务必执行以下操作：

1. 请为 Python 驱动程序完成（先决条件 如果尚未执行此操作）。这包括注册 AWS、授予开发所需的编程访问权限以及安装 Python 3.6 或更高版本。

2. 创建一个名为 quick-start 分类账。

   要了解如何创建分类账，请参阅控制台入门中的 Amazon QLDB 账本的基本操作 或 第 1 步：创建新的分类账。

步骤 1：设置您的项目

首先，请设置您的 Python 项目。

注意

如果您使用IDE具有自动执行这些设置步骤的功能，则可以直接跳到第 2 步：初始化驱动程序。

1. 为应用程序创建一个文件夹。

   $ mkdir myproject
   $ cd myproject

2. 要从 PyPI 安装 Python 的QLDB驱动程序，请输入以下pip命令。

   $ pip install pyqldb

   安装驱动程序还会安装其依赖项，包括 AWS SDK for Python (Boto3) 和 Amazon Ion 软件包。

3. 创建名为 app.py 的新文件。

   然后，按以下步骤逐步添加代码示例，尝试一些基本CRUD操作。或者，您可以跳过本 step-by-step 教程，改为运行完整的应用程序。

第 2 步：初始化驱动程序

初始化连接到名为 quick-start 的分类账的驱动程序实例。将以下代码添加到您的 app.py文件。

from pyqldb.config.retry_config import RetryConfig
from pyqldb.driver.qldb_driver import QldbDriver

# Configure retry limit to 3
retry_config = RetryConfig(retry_limit=3)

# Initialize the driver
print("Initializing the driver")
qldb_driver = QldbDriver("quick-start", retry_config=retry_config)

第 3 步：创建表和索引

以下代码示例显示如何运行 CREATE TABLE 和 CREATE INDEX。

添加以下代码，创建名为 People 的表，并为该表上的 lastName 字段创建索引。索引是优化查询性能和帮助限制乐观并发控制 (OCC) 冲突异常所必需的。

def create_table(transaction_executor):
    print("Creating a table")
    transaction_executor.execute_statement("Create TABLE People")

def create_index(transaction_executor):
    print("Creating an index")
    transaction_executor.execute_statement("CREATE INDEX ON People(lastName)")

# Create a table
qldb_driver.execute_lambda(lambda executor: create_table(executor))

# Create an index on the table
qldb_driver.execute_lambda(lambda executor: create_index(executor))

第 4 步：插入文档

以下代码示例显示如何运行 INSERT 语句。QLDB支持 P artiQL 查询语言（SQL兼容）和 A mazon Ion 数据格式（的超集）。JSON

添加以下代码，在 People 表格中插入文档。

def insert_documents(transaction_executor, arg_1):
    print("Inserting a document")
    transaction_executor.execute_statement("INSERT INTO People ?", arg_1)

# Insert a document
doc_1 = { 'firstName': "John",
          'lastName': "Doe",
          'age': 32,
        }

qldb_driver.execute_lambda(lambda x: insert_documents(x, doc_1))

此示例使用问号（?）作为变量占位符，将文档信息传递给语句。该 execute_statement 方法同时支持 Amazon Ion 类型和 Python 本机类型。

提示

要使用单个 INSERT 语句插入多个文档，可以向该语句传递一个列表类型的参数，如下所示。

# people is a list
transaction_executor.execute_statement("INSERT INTO Person ?", people)

传递 Ion 列表时，不要将变量占位符（?）括在双尖括号（<<...>>）内。在手动 PartiQL 语句中，双尖括号表示名为bag的无序集合。

第 5 步：查询文档

以下代码示例显示如何运行 SELECT 语句。

添加以下代码，用于从 People 表格中查询文档。

def read_documents(transaction_executor):
    print("Querying the table")
    cursor = transaction_executor.execute_statement("SELECT * FROM People WHERE lastName = ?", 'Doe')

    for doc in cursor:
        print(doc["firstName"])
        print(doc["lastName"])
        print(doc["age"])

# Query the table
qldb_driver.execute_lambda(lambda executor: read_documents(executor))

第 6 步：更新文档

以下代码示例显示如何运行 UPDATE 语句。

1. 添加以下代码，通过更新 age 到 42 来更新 People 表中的文档。

   def update_documents(transaction_executor, age, lastName):
       print("Updating the document")
       transaction_executor.execute_statement("UPDATE People SET age = ? WHERE lastName = ?", age, lastName)
   
   # Update the document
   age = 42
   lastName = 'Doe'
   
   qldb_driver.execute_lambda(lambda x: update_documents(x, age, lastName))

2. 再次查询表以查看更新的值。

   # Query the updated document
   qldb_driver.execute_lambda(lambda executor: read_documents(executor))

3. 要运行该应用程序，请在项目目录中输入以下命令。

   $ python app.py

运行完整的应用程序

以下代码示例是 app.py 应用程序的完整版本。您还可以从头到尾复制并运行此代码示例，而不必单独执行前面的步骤。此应用程序演示了对名为的账本的一些基本CRUD操作quick-start。

注意

在运行此代码之前，请确保 quick-start 分类账中还没有名为 People 的活动表。

from pyqldb.config.retry_config import RetryConfig
from pyqldb.driver.qldb_driver import QldbDriver

def create_table(transaction_executor):
    print("Creating a table")
    transaction_executor.execute_statement("CREATE TABLE People")

def create_index(transaction_executor):
    print("Creating an index")
    transaction_executor.execute_statement("CREATE INDEX ON People(lastName)")

def insert_documents(transaction_executor, arg_1):
    print("Inserting a document")
    transaction_executor.execute_statement("INSERT INTO People ?", arg_1)

def read_documents(transaction_executor):
    print("Querying the table")
    cursor = transaction_executor.execute_statement("SELECT * FROM People WHERE lastName = ?", 'Doe')
                                                                                                                                          
    for doc in cursor:
        print(doc["firstName"])
        print(doc["lastName"])
        print(doc["age"])

def update_documents(transaction_executor, age, lastName):
    print("Updating the document")
    transaction_executor.execute_statement("UPDATE People SET age = ? WHERE lastName = ?", age, lastName)

# Configure retry limit to 3
retry_config = RetryConfig(retry_limit=3)

# Initialize the driver
print("Initializing the driver")
qldb_driver = QldbDriver("quick-start", retry_config=retry_config)

# Create a table
qldb_driver.execute_lambda(lambda executor: create_table(executor))

# Create an index on the table
qldb_driver.execute_lambda(lambda executor: create_index(executor))

# Insert a document
doc_1 = { 'firstName': "John",
          'lastName': "Doe",
          'age': 32,
        }

qldb_driver.execute_lambda(lambda x: insert_documents(x, doc_1))

# Query the table
qldb_driver.execute_lambda(lambda executor: read_documents(executor))

# Update the document
age = 42
lastName = 'Doe'

qldb_driver.execute_lambda(lambda x: update_documents(x, age, lastName))

# Query the table for the updated document
qldb_driver.execute_lambda(lambda executor: read_documents(executor))

要运行完整的应用程序，请在项目目录中输入以下命令。

$ python app.py