# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify,
# merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#
# This code expects that you have AWS credentials setup per:
# https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html
from datetime import datetime, timedelta
from logging import basicConfig, getLogger, INFO
from pyqldbsamples.model.sample_data import print_result, get_document_ids, SampleData
from pyqldbsamples.constants import Constants
from pyqldbsamples.connect_to_ledger import create_qldb_driver
logger = getLogger(__name__)
basicConfig(level=INFO)
def format_date_time(date_time):
"""
Format the given date time to a string.
:type date_time: :py:class:`datetime.datetime`
:param date_time: The date time to format.
:rtype: str
:return: The formatted date time.
"""
return date_time.strftime('`%Y-%m-%dT%H:%M:%S.%fZ`')
def previous_primary_owners(driver, vin):
"""
Find previous primary owners for the given VIN in a single transaction.
In this example, query the `VehicleRegistration` history table to find all previous primary owners for a VIN.
:type driver: :py:class:`pyqldb.driver.qldb_driver.QldbDriver`
:param driver: An instance of the QldbDriver class.
:type vin: str
:param vin: VIN to find previous primary owners for.
"""
person_ids = driver.execute_lambda(lambda executor: get_document_ids(executor,
Constants.VEHICLE_REGISTRATION_TABLE_NAME,
'VIN', vin))
todays_date = datetime.utcnow() - timedelta(seconds=1)
three_months_ago = todays_date - timedelta(days=90)
query = 'SELECT data.Owners.PrimaryOwner, metadata.version FROM history({}, {}, {}) AS h WHERE h.metadata.id = ?'.\
format(Constants.VEHICLE_REGISTRATION_TABLE_NAME, format_date_time(three_months_ago),
format_date_time(todays_date))
for ids in person_ids:
logger.info("Querying the 'VehicleRegistration' table's history using VIN: {}.".format(vin))
cursor = driver.execute_lambda(lambda executor: executor.execute_statement(query, ids))
if not (print_result(cursor)) > 0:
logger.info('No modification history found within the given time frame for document ID: {}'.format(ids))
def main(ledger_name=Constants.LEDGER_NAME):
"""
Query a table's history for a particular set of documents.
"""
try:
with create_qldb_driver(ledger_name) as driver:
vin = SampleData.VEHICLE_REGISTRATION[0]['VIN']
previous_primary_owners(driver, vin)
logger.info('Successfully queried history.')
except Exception as e:
logger.exception('Unable to query history to find previous owners.')
raise e
if __name__ == '__main__':
main()
# Copyright 2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.
# SPDX-License-Identifier: MIT-0
#
# Permission is hereby granted, free of charge, to any person obtaining a copy of this
# software and associated documentation files (the "Software"), to deal in the Software
# without restriction, including without limitation the rights to use, copy, modify,
# merge, publish, distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
# INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A
# PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
# HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
# SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
#
# This code expects that you have AWS credentials setup per:
# https://boto3.amazonaws.com/v1/documentation/api/latest/guide/quickstart.html
from datetime import datetime, timedelta
from logging import basicConfig, getLogger, INFO
from pyqldbsamples.model.sample_data import print_result, get_document_ids, SampleData
from pyqldbsamples.constants import Constants
from pyqldbsamples.connect_to_ledger import create_qldb_session
logger = getLogger(__name__)
basicConfig(level=INFO)
def format_date_time(date_time):
"""
Format the given date time to a string.
:type date_time: :py:class:`datetime.datetime`
:param date_time: The date time to format.
:rtype: str
:return: The formatted date time.
"""
return date_time.strftime('`%Y-%m-%dT%H:%M:%S.%fZ`')
def previous_primary_owners(transaction_executor, vin):
"""
Find previous primary owners for the given VIN in a single transaction.
In this example, query the `VehicleRegistration` history table to find all previous primary owners for a VIN.
:type transaction_executor: :py:class:`pyqldb.execution.executor.Executor`
:param transaction_executor: An Executor object allowing for execution of statements within a transaction.
:type vin: str
:param vin: VIN to find previous primary owners for.
"""
person_ids = get_document_ids(transaction_executor, Constants.VEHICLE_REGISTRATION_TABLE_NAME, 'VIN', vin)
todays_date = datetime.utcnow() - timedelta(seconds=1)
three_months_ago = todays_date - timedelta(days=90)
query = 'SELECT data.Owners.PrimaryOwner, metadata.version FROM history({}, {}, {}) AS h WHERE h.metadata.id = ?'.\
format(Constants.VEHICLE_REGISTRATION_TABLE_NAME, format_date_time(three_months_ago),
format_date_time(todays_date))
for ids in person_ids:
logger.info("Querying the 'VehicleRegistration' table's history using VIN: {}.".format(vin))
cursor = transaction_executor.execute_statement(query, ids)
if not (print_result(cursor)) > 0:
logger.info('No modification history found within the given time frame for document ID: {}'.format(ids))
if __name__ == '__main__':
"""
Query a table's history for a particular set of documents.
"""
try:
with create_qldb_session() as session:
vin = SampleData.VEHICLE_REGISTRATION[0]['VIN']
session.execute_lambda(lambda lambda_executor: previous_primary_owners(lambda_executor, vin),
lambda retry_attempt: logger.info('Retrying due to OCC conflict...'))
logger.info('Successfully queried history.')
except Exception:
logger.exception('Unable to query history to find previous owners.')
