Configurar la función Advanced I/O
A continuación, comenzaremos a codificar la aplicación de lista de tareas configurando el componente de función.
El directorio de la función, functions/to_do_list_function contiene:
-
El archivo principal de la función main.py
-
El archivo de configuración catalyst-config.json
-
El archivo requirements.txt para mencionar cualquier biblioteca externa que pueda agregar.
Agregará código en el archivo main.py.
Las tres APIs en la función Advanced I/O que manejan el enrutamiento entre el servidor y el Data Store son:
-
GET /todo: Para obtener los elementos de la lista de tareas de la tabla TodoItems en el Data Store
-
POST /todo: Para crear y guardar un nuevo elemento de la lista en el Data Store
-
DELETE /todo: Para eliminar un elemento de la lista del Data Store
A continuación, agreguemos el código en el archivo de la función. Copie el código Python y péguelo en main.py en el directorio functions/to_do_list_function de su proyecto, y guarde el archivo. Puede usar cualquier IDE de su preferencia para trabajar con los archivos de la aplicación.
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']['COUNT(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': 'Proporcione un rowid válido en la ruta de la solicitud'
}
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": "Ocurrió un error interno del servidor. Intente nuevamente en un momento."
}), 500)
return response
El directorio de funciones está ahora configurado.
Última actualización 2026-03-20 21:51:56 +0530 IST