¡Hola, amigos!
Últimamente, he estado probando diferentes cosas para desarrollar muchas más aplicaciones de utilidad que se conectan a la blockchain y que las genero a través con Python y la Liberia Beem, así durante las experimentaciones he aprendido un poco de cosas, pero también con la ayuda de AI en algunas cosas.
El experimento que he realizado hoy es probar una manera de balancear la carga de las peticiones que hago a las API para no sobrecargar el nodo, así que quiero cada 1000 peticiones se cambie la API.
Así que implementé lo siguiente:
nodes = [
"https://anyx.io",
"https://api.openhive.network",
"https://hive-api.3speak.tv",
"https://rpc.ausbit.dev",
"https://api.deathwing.me"
]
def get_next_api_url():
node_list = nodes
while True:
for api_url in node_list:
yield api_url
next_api_url = get_next_api_url()
Luego procedí a visualizar como puedo extraer una nueva API, así que lo que me funcionó fue colocar esto en donde se hagan las peticiones.
if api_calls >= 1000:
print ("Alcanzado los 1000")
api_calls = 0
switch_hive_node()
Cuando llega a las 1000 peticiones, me notifica:
Alcanzado los 1000
Switching to new Hive API node: https://api.openhive.network
También ajuste lo siguiente:
hive_instance = Hive(node=api_url, num_retries=1, num_retries_call=1, timeout=1)
num_retries=1, num_retries_call=1 y timeout=1, todos los tengo en 1 para que no tarde mucho en cambiar de API si genera un error.
Probando
La función para probar este sistema de rotar las API cada 1000 peticiones:
def get_hive_posts(start_date):
hive = switch_hive_node()
while True:
try:
blockchain = Blockchain(blockchain_instance=hive)
start_block = blockchain.get_estimated_block_num(start_date)
end_block = blockchain.get_estimated_block_num(datetime.utcnow())
break
except Exception as e:
print(f"Error: {e}. Retrying in 5 seconds...")
time.sleep(5)
print(start_block)
print(end_block)
posts = {}
api_calls = 0
for op in blockchain.stream(opNames=["comment"], start=start_block, stop=end_block, threading=True, thread_num=8):
if api_calls >= 1000:
print ("Alcanzado los 1000")
api_calls = 0
switch_hive_node()
if op["parent_author"] == "":
print (op["trx_id"])
print (op["author"])
try:
json_metadata = json.loads(op["json_metadata"])
except (json.JSONDecodeError, TypeError):
continue
tags = []
if isinstance(json_metadata, dict) and "tags" in json_metadata:
tags = json_metadata["tags"]
elif isinstance(json_metadata, list):
tags = json_metadata
if isinstance(tags, list) and "spanish" in tags:
post_data = {
"author": op["author"],
"url": f"https://peakd.com/@{op['author']}/{op['permlink']}"
}
post_key = f"{op['author']}_{op['permlink']}"
if post_key not in posts or posts[post_key]['block_num'] < op['block_num']:
post_data['block_num'] = op['block_num']
posts[post_key] = post_data
api_calls += 1
return list(posts.values())
if __name__ == "__main__":
posts = get_hive_posts(start_date)
table = PrettyTable()
table.field_names = ["Usuario", "URL (peakd.com)"]
for post in posts:
table.add_row([post["author"], post["url"]])
print(table)
Aquí parte de los resultados de los post publicados con el tag Spanish
Sin duda extraer datos de este tipo donde se explora cada bloque es mejor usar hiveSQL porque es mucho más rápido, pero se puede considerar tener nuestra propia basa de datos SQL en el caso de crear una App que necesite extraer datos de la blockchain.
En mi próxima publicación voy a tratar de extraer el top 15 de autores de la semana mediante este método.
¿Te gustaría aprender a crear aplicaciones en Hive?
También estoy dispuesto a recibir sugerencias