Controlador Amazon QLDB para Python: libro de recetas de referencia - Amazon Quantum Ledger Database (Amazon QLDB)

Controlador Amazon QLDB para Python: libro de recetas de referencia

Esta guía de referencia muestra los casos de uso más comunes del controlador Amazon QLDB para Python. En él se proporcionan ejemplos de código Python que muestran cómo utilizar el controlador para ejecutar operaciones básicas de creación, lectura, actualización y eliminación (CRUD, por sus siglas en inglés). También incluye ejemplos de código para procesar datos de Amazon Ion. Además, esta guía destaca las mejores prácticas para hacer que las transacciones sean idempotentes e implementar restricciones de exclusividad.

nota

Cuando proceda, algunos pasos incluyen ejemplos de código diferentes para cada versión principal compatible del controlador QLDB para Python.

Importación del controlador

En los ejemplos de código se utilizan las siguientes importaciones.

3.x
from pyqldb.driver.qldb_driver import QldbDriver import amazon.ion.simpleion as simpleion
2.x
from pyqldb.driver.pooled_qldb_driver import PooledQldbDriver import amazon.ion.simpleion as simpleion
nota

En este ejemplo también se importa el paquete Amazon Ion (amazon.ion.simpleion). Necesita este paquete para procesar los datos de Ion al ejecutar algunas operaciones de datos de esta referencia. Para obtener más información, consulte Trabajar con Amazon Ion.

Instanciación del controlador

En el siguiente ejemplo de código, se crea una instancia del controlador que se conecta a un nombre de libro mayor especificado mediante la configuración predefinida.

3.x
qldb_driver = QldbDriver(ledger_name='vehicle-registration')
2.x
qldb_driver = PooledQldbDriver(ledger_name='vehicle-registration')

Operaciones CRUD

La QLDB ejecuta operaciones de creación, lectura, actualización y eliminación (CRUD, por sus siglas en inglés) como parte de una transacción.

aviso

Como práctica recomendada, haga que sus transacciones de escritura sean estrictamente idempotentes.

Hacer que las transacciones sean idempotentes

Le recomendamos que haga que las transacciones de escritura sean idempotentes para evitar cualquier efecto secundario inesperado en caso de reintentos. Una transacción es idempotente si puede ejecutarse varias veces y producir resultados idénticos cada vez.

Por ejemplo, pensemos en una transacción que inserta un documento en una tabla llamada Person. La transacción debe comprobar primero si el documento ya existe o no en la tabla. Sin esta comprobación, la tabla podría terminar con documentos duplicados.

Supongamos que la QLDB confirma correctamente la transacción en el lado del servidor pero el cliente agota el tiempo de espera mientras espera una respuesta. Si la transacción no es idempotente, se podría insertar el mismo documento más de una vez en caso de volver a intentarlo.

Uso de índices para evitar escanear tablas completas

También le recomendamos que ejecute instrucciones con una frase de predicado WHERE utilizando un operador de igualdad sobre un campo indexado o un ID de documento; por ejemplo, WHERE indexedField = 123 o WHERE indexedField IN (456, 789). Sin esta búsqueda indexada, la QLDB necesita escanear las tablas, lo que puede provocar tiempos de espera de las transacciones o conflictos de control de concurrencia optimista (OCC).

Para obtener más información acerca de OCC, consulte Modelo de concurrencia de Amazon QLDB.

Transacciones creadas de forma implícita

El método pyqldb.driver.qldb_driver.execute_lambda acepta una función de Lambda que recibe una instancia de pyqldb.execution.executor.Executor, que se puede utilizar para ejecutar instrucciones. La instancia de Executor envuelve una transacción creada de forma implícita.

Puede ejecutar instrucciones dentro de la función de Lambda mediante el método execute_statement de transacciones. El controlador confirma implícitamente la transacción cuando vuelve la función de Lambda.

nota

El método execute_statement admite valores tanto en los tipos de Amazon Ion como en los tipos nativos de Python. Si pasa un tipo nativo de Python como argumento a execute_statement, el controlador lo convierte en un tipo Ion mediante el paquete amazon.ion.simpleion(siempre que se admita la conversión para el tipo de datos Python indicado). Para conocer los tipos de datos y las reglas de conversión compatibles, consulte el código origen de Ion simple.

En las siguientes secciones se muestra cómo ejecutar operaciones CRUD básicas, especificar una lógica de reintento personalizada e implementar restricciones de exclusividad.

Creación de tablas

def create_table(transaction_executor): transaction_executor.execute_statement("CREATE TABLE Person") qldb_driver.execute_lambda(lambda executor: create_table(executor))

Creación de índices

def create_index(transaction_executor): transaction_executor.execute_statement("CREATE INDEX ON Person(GovId)") qldb_driver.execute_lambda(lambda executor: create_index(executor))

Lectura de documentos

# Assumes that Person table has documents as follows: # { "GovId": "TOYENC486FH", "FirstName": "Brent" } def read_documents(transaction_executor): cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = 'TOYENC486FH'") for doc in cursor: print(doc["GovId"]) # prints TOYENC486FH print(doc["FirstName"]) # prints Brent qldb_driver.execute_lambda(lambda executor: read_documents(executor))

Uso de parámetros de consulta

En el siguiente ejemplo de código, se utiliza un parámetro de consulta de tipo nativo.

cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = ?", 'TOYENC486FH')

En el siguiente ejemplo de código, se utiliza un parámetro de consulta de tipo Ion.

name = ion.loads('Brent') cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE FirstName = ?", name)

En el siguiente ejemplo de código se utilizan varios parámetros de consulta.

cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = ? AND FirstName = ?", 'TOYENC486FH', "Brent")

En el siguiente ejemplo de código, se utiliza una lista de parámetros de consulta.

gov_ids = ['TOYENC486FH','ROEE1','YH844'] cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId IN (?,?,?)", *gov_ids)
nota

Cuando ejecuta una consulta sin una búsqueda indexada, se invoca un escaneo completo de la tabla. En este ejemplo, se recomienda tener un índice en el campo GovId para optimizar el rendimiento. Sin un índice en GovId, las consultas pueden tener más latencia y, además, provocar excepciones de conflictos de OCC o tiempos de espera de las transacciones.

Inserción de documentos

El siguiente ejemplo de código inserta los tipos de datos nativos.

def insert_documents(transaction_executor, arg_1): # Check if doc with GovId:TOYENC486FH exists # This is critical to make this transaction idempotent cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = ?", 'TOYENC486FH') # Check if there is any record in the cursor first_record = next(cursor, None) if first_record: # Record already exists, no need to insert pass else: transaction_executor.execute_statement("INSERT INTO Person ?", arg_1) doc_1 = { 'FirstName': "Brent", 'GovId': 'TOYENC486FH', } qldb_driver.execute_lambda(lambda executor: insert_documents(executor, doc_1))

El siguiente ejemplo de código inserta los tipos de datos de Ion.

def insert_documents(transaction_executor, arg_1): # Check if doc with GovId:TOYENC486FH exists # This is critical to make this transaction idempotent cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = ?", 'TOYENC486FH') # Check if there is any record in the cursor first_record = next(cursor, None) if first_record: # Record already exists, no need to insert pass else: transaction_executor.execute_statement("INSERT INTO Person ?", arg_1) doc_1 = { 'FirstName': 'Brent', 'GovId': 'TOYENC486FH', } # create a sample Ion doc ion_doc_1 = simpleion.loads(simpleion.dumps(doc_1))) qldb_driver.execute_lambda(lambda executor: insert_documents(executor, ion_doc_1))

Esta transacción inserta un documento en la tabla Person. Antes de insertar, compruebe primero si el documento ya existe en la tabla. Esta comprobación hace que la transacción sea de naturaleza idempotente. Incluso si realiza esta transacción varias veces, no provocará ningún efecto secundario no deseado.

nota

En este ejemplo, se recomienda tener un índice en el campo GovId para optimizar el rendimiento. Sin un índice en GovId, las instrucciones pueden tener más latencia y, además, provocar excepciones de conflictos de OCC o tiempos de espera de las transacciones.

Insertar varios documentos en una instrucción

Para insertar varios documentos mediante una sola instrucción INSERT, puede pasar un parámetro del tipo list a la instrucción de la siguiente manera.

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

No coloque el marcador de posición variable (?) entre corchetes de doble ángulo (<<...>>) al pasar una lista. En las instrucciones PartiQL manuales, los corchetes de doble ángulo indican una colección desordenada conocida como bolsa.

Actualización de documentos

El siguiente ejemplo de código utiliza tipos de datos nativos.

def update_documents(transaction_executor, gov_id, name): transaction_executor.execute_statement("UPDATE Person SET FirstName = ? WHERE GovId = ?", name, gov_id) gov_id = 'TOYENC486FH' name = 'John' qldb_driver.execute_lambda(lambda executor: update_documents(executor, gov_id, name))

El siguiente ejemplo de código utiliza tipos de datos Ion.

def update_documents(transaction_executor, gov_id, name): transaction_executor.execute_statement("UPDATE Person SET FirstName = ? WHERE GovId = ?", name, gov_id) # Ion datatypes gov_id = simpleion.loads('TOYENC486FH') name = simpleion.loads('John') qldb_driver.execute_lambda(lambda executor: update_documents(executor, gov_id, name))
nota

En este ejemplo, se recomienda tener un índice en el campo GovId para optimizar el rendimiento. Sin un índice en GovId, las instrucciones pueden tener más latencia y, además, provocar excepciones de conflictos de OCC o tiempos de espera de las transacciones.

Eliminación de documentos

El siguiente ejemplo de código utiliza tipos de datos nativos.

def delete_documents(transaction_executor, gov_id): cursor = transaction_executor.execute_statement("DELETE FROM Person WHERE GovId = ?", gov_id) gov_id = 'TOYENC486FH' qldb_driver.execute_lambda(lambda executor: delete_documents(executor, gov_id))

El siguiente ejemplo de código utiliza tipos de datos Ion.

def delete_documents(transaction_executor, gov_id): cursor = transaction_executor.execute_statement("DELETE FROM Person WHERE GovId = ?", gov_id) # Ion datatypes gov_id = simpleion.loads('TOYENC486FH') qldb_driver.execute_lambda(lambda executor: delete_documents(executor, gov_id))
nota

En este ejemplo, se recomienda tener un índice en el campo GovId para optimizar el rendimiento. Sin un índice en GovId, las instrucciones pueden tener más latencia y, además, provocar excepciones de conflictos de OCC o tiempos de espera de las transacciones.

Ejecutar varias instrucciones en una transacción

# This code snippet is intentionally trivial. In reality you wouldn't do this because you'd # set your UPDATE to filter on vin and insured, and check if you updated something or not. def do_insure_car(transaction_executor, vin): cursor = transaction_executor.execute_statement( "SELECT insured FROM Vehicles WHERE vin = ? AND insured = FALSE", vin) first_record = next(cursor, None) if first_record: transaction_executor.execute_statement( "UPDATE Vehicles SET insured = TRUE WHERE vin = ?", vin) return True else: return False def insure_car(qldb_driver, vin_to_insure): return qldb_driver.execute_lambda( lambda executor: do_insure_car(executor, vin_to_insure))

Lógica de reintentos

El método del controlador execute_lambda tiene un mecanismo de reintento integrado que reintenta la transacción si se produce una excepción que se pueda volver a intentar (como tiempos de espera o conflictos de OCC).

3.x

El número máximo de reintentos y la estrategia de retardo se pueden configurar.

El límite de reintentos predeterminado es 4 y la estrategia de retardo predeterminada es retroceso exponencial y fluctuación, con una base de 10 milisegundos. Puede establecer la configuración de reintentos por instancia de controlador y también por transacción mediante una instancia de pyqldb.config.retry_config.RetryConfig.

El siguiente ejemplo de código especifica la lógica de reintentos con un límite de reintentos personalizado y una estrategia de retardo personalizada para una instancia del controlador.

from pyqldb.config.retry_config import RetryConfig from pyqldb.driver.qldb_driver import QldbDriver # Configuring retry limit to 2 retry_config = RetryConfig(retry_limit=2) qldb_driver = QldbDriver("test-ledger", retry_config=retry_config) # Configuring a custom backoff which increases delay by 1s for each attempt. def custom_backoff(retry_attempt, error, transaction_id): return 1000 * retry_attempt retry_config_custom_backoff = RetryConfig(retry_limit=2, custom_backoff=custom_backoff) qldb_driver = QldbDriver("test-ledger", retry_config=retry_config_custom_backoff)

El siguiente ejemplo de código especifica la lógica de reintentos con un límite de reintentos personalizado y una estrategia de retardo personalizada para una ejecución lambda particular. Esta configuración execute_lambda anula la lógica de reintentos establecida para la instancia del controlador.

from pyqldb.config.retry_config import RetryConfig from pyqldb.driver.qldb_driver import QldbDriver # Configuring retry limit to 2 retry_config_1 = RetryConfig(retry_limit=4) qldb_driver = QldbDriver("test-ledger", retry_config=retry_config_1) # Configuring a custom backoff which increases delay by 1s for each attempt. def custom_backoff(retry_attempt, error, transaction_id): return 1000 * retry_attempt retry_config_2 = RetryConfig(retry_limit=2, custom_backoff=custom_backoff) # The config `retry_config_1` will be overriden by `retry_config_2` qldb_driver.execute_lambda(lambda txn: txn.execute_statement("CREATE TABLE Person"), retry_config_2)
2.x

El número máximo de reintentos se puede configurar. Puede configurar el límite de reintentos estableciendo la propiedad retry_limit al inicializar PooledQldbDriver.

El límite de reintentos predeterminado es de 4.

Implementación de restricciones de exclusividad

QLDB no admite índices únicos, pero puede implementar este comportamiento en su aplicación.

Suponga que desea implementar una restricción de exclusividad en el campo GovId de la tabla Person. Para ello, puede escribir una transacción que haga lo siguiente:

  1. Afirmar que en la tabla no hay documentos existentes con un GovId especificado.

  2. Insertar el documento si se aprueba la afirmación.

Si una transacción competidora supera la afirmación simultáneamente, solo una de las transacciones se confirmará correctamente. La otra transacción fallará y se producirá una excepción de conflicto de OCC.

En el siguiente ejemplo de código, se muestra cómo implementar esta lógica de restricción de exclusividad.

def insert_documents(transaction_executor, gov_id, document): # Check if doc with GovId = gov_id exists cursor = transaction_executor.execute_statement("SELECT * FROM Person WHERE GovId = ?", gov_id) # Check if there is any record in the cursor first_record = next(cursor, None) if first_record: # Record already exists, no need to insert pass else: transaction_executor.execute_statement("INSERT INTO Person ?", document) qldb_driver.execute_lambda(lambda executor: insert_documents(executor, gov_id, document))
nota

En este ejemplo, se recomienda tener un índice en el campo GovId para optimizar el rendimiento. Sin un índice en GovId, las instrucciones pueden tener más latencia y, además, provocar excepciones de conflictos de OCC o tiempos de espera de las transacciones.

Trabajar con Amazon Ion

Las siguientes secciones muestran cómo usar el módulo Amazon Ion para procesar los datos de Ion.

Importación del módulo Ion

import amazon.ion.simpleion as simpleion

Creación de tipos de Ion

El siguiente ejemplo de código crea un objeto Ion a partir del texto de Ion.

ion_text = '{GovId: "TOYENC486FH", FirstName: "Brent"}' ion_obj = simpleion.loads(ion_text) print(ion_obj['GovId']) # prints TOYENC486FH print(ion_obj['Name']) # prints Brent

El siguiente ejemplo de código crea un objeto Ion a partir de un dict de Python.

a_dict = { 'GovId': 'TOYENC486FH', 'FirstName': "Brent" } ion_obj = simpleion.loads(simpleion.dumps(a_dict)) print(ion_obj['GovId']) # prints TOYENC486FH print(ion_obj['FirstName']) # prints Brent

Obtener un volcado binario de Ion

# ion_obj is an Ion struct print(simpleion.dumps(ion_obj)) # b'\xe0\x01\x00\xea\xee\x97\x81\x83\xde\x93\x87\xbe\x90\x85GovId\x89FirstName\xde\x94\x8a\x8bTOYENC486FH\x8b\x85Brent'

Obtener un volcado de texto de Ion

# ion_obj is an Ion struct print(simpleion.dumps(ion_obj, binary=False)) # prints $ion_1_0 {GovId:'TOYENC486FH',FirstName:"Brent"}

Para obtener más información acerca de cómo trabajar con Ion, consulte la documentación de Amazon Ion en GitHub. Para ver más ejemplos de código sobre cómo trabajar con Ion en QLDB, consulte Uso de tipos de datos de Amazon Ion en Amazon QLDB.