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.
main.pycopyimport 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 +0530