Configure the Advanced I/O Function

Next, we will begin coding the to-do list application by configuring the function component.

The function’s directory, functions/to_do_list_function contains:

The main.py main function file
The catalyst-config.json configuration file
The requirements.txt file to mention any external libraries that you might add.

You will be adding code in the main.py file.

The three APIs in the Advanced I/O function that handle the routing between the server and the Data Store are:

GET /todo: To obtain the to-do list items from the TodoItems table in the Data Store
POST /todo: To create and save a new list item in the Data Store
DELETE /todo: To delete a list item from the Data Store

Next, let’s add the code in the function file. Copy the Python code and paste it in main.py in the functions/to_do_list_function directory of your project, and save the file. You can use any IDE of your choice to work with the application’s files.

Note: Please go through the code given in this section to ensure you fully understand it. We will discuss the architecture of the function and the client in the next section.

main.py
copy

import json
import zcatalyst_sdk
import logging
from flask import Request, make_response, jsonify
logger = logging.getLogger()
def handler(request: Request):
try:
app = zcatalyst_sdk.initialize()
logger = logging.getLogger()
if request.path == “/add” and request.method == ‘POST’:
req_data = request.get_json()
notes = req_data.get(“notes”)        

table = app.datastore().table(‘TodoItems’)
row = table.insert_row({
‘Notes’: notes
})
todo_item = {‘id’: row[‘ROWID’], ’notes’: notes}     

response_data = {
‘status’: ‘success’,
‘data’: {
’todoItem’: todo_item
}
}
return make_response(jsonify(response_data), 200)
elif request.path == “/all” and request.method == ‘GET’:
page = request.args.get(‘page’)
per_page = request.args.get(‘perPage’)
page = int(page)
logger.info(page)
per_page = int(per_page)
logger.info(per_page)
zcql_service = app.zcql()  

row_count = zcql_service.execute_query(‘SELECT COUNT(ROWID) FROM TodoItems’)
row_id = int(row_count[0][‘TodoItems’][‘ROWID’])
has_more = int(row_id) > (page) * (per_page)
logger.info(has_more)
query_result = zcql_service.execute_query(f’SELECT ROWID, Notes FROM TodoItems LIMIT {(page - 1) * per_page + 1},{per_page}’)
todo_items = [{‘id’: item[‘TodoItems’][‘ROWID’], ’notes’: item[‘TodoItems’][‘Notes’]} for item in query_result]
get_resp = {
‘status’: ‘success’,
‘data’: {
’todoItems’: todo_items,
‘hasMore’: has_more
}
}
return make_response(jsonify(get_resp), 200)
elif request.method == ‘DELETE’:
row_id = request.path[1:]
if row_id:
datastore_service = app.datastore()
table_service = datastore_service.table(“ToDoItems”)
table_service.delete_row(row_id)
row_response = {
‘status’: ‘success’,
‘data’: {
’todoItem’: {
‘id’: row_id
}
}}      

logging.info(row_response)
return make_response(jsonify(row_response), 200)
else:
message = {
‘status’: ‘failure’,
’error’: ‘Please provide a valid rowid in the request path’
}
return make_response(jsonify(message), 400)
else:
print(‘working’)
except Exception as err:
logger.error(f"Exception in to_do_list_function :{err}")
response = make_response(jsonify({
“error”: “Internal server error occurred. Please try again in some time.”
}), 500)
return response

The functions directory is now configured.

Last Updated 2023-12-14 16:25:23 +0530 IST

Advanced I/O Functions Functions Directory Structure