Configurar la Job Function
A continuación, comenzaremos a codificar la aplicación de noticias configurando el componente de la job function.
El directorio de la función, en este caso functions/NewsApp, contiene:
- El archivo principal de la función main.py
- El archivo de configuración catalyst-config.json
- El archivo requirements.txt
Agregaremos código en el archivo main.py. No modificará el código de los archivos de configuración y dependencias.
Como se mencionó en la introducción, la job function realiza dos tareas: hacer las llamadas API a NewsAPI para obtener noticias, y poblar las noticias en Catalyst Data Store. Las llamadas API se realizan utilizando una clave API proporcionada por NewsAPI.
Registrarse en NewsAPI
Antes de codificar la job function, debe registrarse para una suscripción gratuita de desarrollador con NewsAPI y obtener la clave API de la siguiente manera:
-
Visite https://newsapi.org/register.
-
Proporcione los detalles requeridos y haga clic en Submit.
Agregar código de la función
Puede copiar el código a continuación y pegarlo en main.py ubicado en el directorio functions/NewsApp de su proyecto y guardar el archivo. Puede usar cualquier IDE para trabajar con los archivos de la aplicación.
import logging
import zcatalyst_sdk
import requests
logger = logging.getLogger()
def handler(job_request, context):
logger.info("Fetching news!")
# Los nombres de las tablas en el Data Store donde se almacenarán los elementos de noticias
table_name = ["HEADLINES", "BUSINESS", "ENTERTAINMENT", "HEALTH", "SCIENCE", "SPORTS", "TECHNOLOGY"]
country = "US" #Obtiene los elementos de noticias de los Estados Unidos de América
api_key = "cc30ddcaxxxxxxxxxx" # Proporcione la clave API que obtuvo de NewsAPI dentro de las comillas
fetch_url = "https://newsapi.org/v2/top-headlines"
#Obtiene las noticias de última hora de diferentes categorías con sus titulares
for category in table_name:
# Construyendo lo requerido para obtener noticias
params = {
'country':country,
'apiKey':api_key,
}
if category != "HEADLINES":
params['category'] = category
# Realiza una llamada API a News API para obtener los datos
response = requests.get(fetch_url, params=params)
# Si la respuesta es 200, los datos se mueven al Data Store. Si no, el error se registra.
if response.status_code == 200:
data = response.json()
try:
if 'articles' in data:
array_data = data['articles'][:15]
logger.info("15 days "+str(array_data))
# Este método inserta/actualiza las noticias en el Data Store
push_to_datastore(array_data,category)
except requests.RequestException as e:
logger.error("Request failed: "+ str(e))
context.close_with_failure() # fin de la aplicación con error
# Las acciones se registran. Puede verificar los logs desde Catalyst Logs.
logger.info("New updated")
else:
logger.error("Request failed with status :: ",response.status_code)
context.close_with_failure() # fin de la aplicación con error
context.close_with_success() # fin de la aplicación con éxito
def push_to_datastore(sliced_data,tableName):
app = zcatalyst_sdk.initialize()
zcql_service = app.zcql()
datastore_service = app.datastore()
#Define la consulta ZCQL que se usará para encontrar el número de filas en una tabla
query = "select ROWID from "+tableName
zcql_result = zcql_service.execute_query(query)
table_instance = datastore_service.table(tableName)
#Inserta / Actualiza los datos obtenidos de la API en el datastore
for i in range(0,14):
title = sliced_data[i]['title']
url = sliced_data[i]['url']
if len(zcql_result) == 0:
# insertando nueva fila
row_data= {'title':title,'url':url}
table_instance.insert_row(row_data)
logger.info("Row inserted "+ str(row_data))
else:
# actualizando fila
row_data= {'title':title,'url':url,'ROWID':zcql_result[i][tableName]['ROWID']}
table_instance.update_row(row_data)
logger.info("Row inserted "+ str(row_data))
La job function está ahora configurada. Discutiremos la arquitectura de la aplicación después de configurar el client.
Última actualización 2026-03-20 21:51:56 +0530 IST