El asincronismo en Python te permite ejecutar varias tareas al mismo tiempo sin que tu programa se quede esperando. Aprenderás a usar asyncio, corrutinas, async y await para acortar tiempos de procesamiento y responder a los usuarios de forma más eficiente. Es ideal si trabajas con APIs, bases de datos o servidores que tardan en responder.
¿Qué diferencia hay entre sincronismo y asincronismo?
Cuando trabajas de forma sincrónica, envías una petición al servidor, esperas la respuesta y recién entonces puedes enviar la siguiente. Es un flujo lineal que bloquea el programa mientras espera.
Con asincronismo, en cambio, puedes enviar múltiples tareas al mismo tiempo. El servidor las procesa y devuelve las respuestas dentro del mismo tiempo de ejecución. Y aquí viene lo interesante: mientras una tarea está pausada esperando, otras pueden avanzar.
¿Cuándo conviene usar asincronismo? Cuando tu programa depende de respuestas externas como servidores, APIs o bases de datos. Así evitas que una tarea lenta bloquee a las demás.
¿Qué son las corrutinas y cómo funcionan con async y await?
En Python, el asincronismo se apoya en las corrutinas, que son funciones que pueden ser pausadas y luego retomadas. Esa pausa es lo que permite que otras tareas se ejecuten mientras una espera.
Para declararlas, usas dos palabras reservadas:
async: define que una función es asíncrona, es decir, una corrutina.
await: marca el punto donde la corrutina se pausa hasta que termine la operación que está esperando.
La librería que orquesta todo esto es asyncio, encargada de crear un event loop que recibe los procesos, los administra y devuelve la respuesta de cada tarea.
¿Cómo implementar asyncio paso a paso en Python?
Vamos a construir un ejemplo donde simulamos el procesamiento de un archivo. Primero, importas la librería y defines tu primera corrutina [00:54].
Definir la corrutina que procesa datos
python
import asyncio
async def process_data(data):
print(f"Procesando {data}...")
await asyncio.sleep(10)
print(f"{data} ya terminó de procesarse")
return data * 2
Aquí pasan varias cosas clave. La función está marcada con async, lo que la convierte en corrutina. Dentro usamos await asyncio.sleep(10) para simular una operación que tarda 10 segundos sin bloquear el event loop [01:48]. Y al final retornamos el dato multiplicado por dos.
¿Qué hace exactamente await? Pausa la corrutina en ese punto, libera el event loop para que otras tareas avancen y retoma la ejecución cuando la operación termina.
Crear la función principal y arrancar el event loop
Ahora necesitas una corrutina principal que coordine las llamadas:
La función main también es asíncrona y desde ahí llamas a process_data. Para iniciar todo, usas asyncio.run(main()), que arranca el event loop y ejecuta las corrutinas [03:30].
¿Qué hace el event loop al ejecutar las corrutinas?
El event loop es el motor que coordina toda la asincronía. Cuando lanzas asyncio.run, el ciclo arranca con el primer request y administra cada corrutina hasta que devuelve su resultado.
En el ejemplo, el flujo es así:
Inicia el procesamiento del archivo TXT.
La corrutina se pausa 10 segundos por el await asyncio.sleep.
Durante esa pausa, otras corrutinas podrían ejecutarse.
Al terminar, el event loop devuelve el resultado y muestra el mensaje final.
En este ejercicio simulamos el tiempo de espera, pero en proyectos reales esa pausa la define el servidor, la API o la base de datos que estés consultando.
Reto: gestor de descarga de archivos con tiempos aleatorios
Para afianzar lo aprendido, crea un sistema de gestión de descargas donde cada archivo tenga un tiempo de procesamiento diferente y aleatorio. En el ejemplo de la clase fijamos 10 segundos manualmente; ahora la idea es que uses valores variables para simular descargas reales.
Algunas pistas para resolverlo:
Usa el módulo random para generar tiempos distintos por archivo.
Define una corrutina por descarga y lánzalas en paralelo desde main.
Observa cómo el event loop maneja varias corrutinas pausadas al mismo tiempo.
Dominar el asincronismo te ayudará a construir proyectos más eficientes, porque puedes pausar una tarea mientras esperas una respuesta y aprovechar ese tiempo para ejecutar otras. ¿Qué tipo de proyecto te gustaría optimizar con asyncio? Cuéntame en los comentarios.
import asyncio
import random
async def donwload_process(file):print(f'Descargando archivo {file}') time=random.randint(1,10)await asyncio.sleep(time)print(f'{file} downloaded')return file
async def main():print('Inicio de la descarga') result=awaitdonwload_process('file1.txt') result1=awaitdonwload_process('file2.txt') result2=awaitdonwload_process('file3.txt')print(f'{result,result1,result2}')asyncio.run(main())
muy buen aporte
Gracia por tu aporte
El asincronismo en Python permite ejecutar tareas concurrentes sin bloquear el flujo de ejecución del programa. La biblioteca **asyncio** de Python facilita la creación y gestión de tareas asíncronas, lo que es ideal para aplicaciones que necesitan manejar múltiples operaciones de entrada y salida (I/O) de manera eficiente, como las solicitudes de red, el acceso a bases de datos y la lectura de archivos.
### Fundamentos del Asincronismo con asyncio
- **Asincronismo**: Consiste en realizar tareas de manera concurrente, de modo que una tarea pueda comenzar mientras otra está en pausa, sin detener el flujo general del programa.
- **async y await**: Estas palabras clave son esenciales en asyncio. async se usa para definir una función asíncrona (async def), mientras que await se usa dentro de estas funciones para pausar su ejecución hasta que se complete una operación asíncrona.
- **Corutinas**: Las funciones asíncronas en Python son "corutinas", que pueden ser pausadas y reanudadas. Esto permite que otras corutinas se ejecuten mientras una está esperando, optimizando el rendimiento.
### Ejemplo Básico con asyncio
Supongamos que tenemos una tarea simple que se ejecuta durante algunos segundos, como simulación de una tarea de red o una solicitud a una API.
import asyncio
asyncdeftarea(nombre, duracion):  print(f"Iniciando {nombre}")  await asyncio.sleep(duracion) # Simula una tarea que toma tiempo  print(f"Terminando {nombre} después de {duracion} segundos")
\# Ejecutar varias tareas
asyncdefmain():  await asyncio.gather(  tarea("Tarea 1", 2),  tarea("Tarea 2", 3),  tarea("Tarea 3", 1)  )
\# Ejecutar la función principal
asyncio.run(main())
### Explicación
- async def tarea(nombre, duracion): Define una función asíncrona que simula el trabajo con asyncio.sleep(duracion).
- await asyncio.gather(...): Ejecuta varias tareas de forma concurrente. gather acepta múltiples corutinas y las ejecuta al mismo tiempo, permitiendo que las tareas esperen sin bloquear el flujo general del programa.
### Uso de asyncio.gather y asyncio.sleep
- **asyncio.gather**: Permite agrupar varias corutinas para ejecutarlas al mismo tiempo. Esto es útil cuando tienes varias tareas independientes que deseas que se completen en paralelo.
- **asyncio.sleep**: Simula una operación de espera asíncrona, como el tiempo que toma una solicitud a una API o la lectura de un archivo.
### Ejemplo Realista: Descarga Simultánea de Páginas Web
Imaginemos un programa que descarga contenido de varias URLs. Podemos simular el proceso de espera de las respuestas de una API con asyncio.sleep.
import asyncio
asyncdef descargar\_pagina(url):  print(f"Descargando {url}")  await asyncio.sleep(2) # Simula la espera de respuesta de la red  print(f"Descarga completa: {url}")  return f"Contenido de {url}"asyncdefmain():  urls = \["https://ejemplo.com/1", "https://ejemplo.com/2", "https://ejemplo.com/3"]     \# Ejecutar todas las descargas en paralelo  resultados = await asyncio.gather(\*(descargar\_pagina(url) for url in urls))     print("Resultados de las descargas:")  for resultado in resultados:  print(resultado)asyncio.run(main())
### Explicación
- **Lista de URLs**: Simulamos una lista de URLs que queremos "descargar".
- **await asyncio.gather(...)**: Llama a descargar\_pagina para cada URL en paralelo, esperando los resultados de todas las descargas antes de continuar.
### Ejecución Secuencial vs. Asíncrona
Si las descargas se ejecutaran de forma secuencial, cada una tendría que esperar a que la anterior termine. Con asyncio.gather, se ejecutan en paralelo, y el tiempo total es aproximadamente el de la tarea más larga.
### Buenas Prácticas en asyncio
1. **Usa await correctamente**: Solo se puede utilizar await dentro de funciones definidas con async def. Usar await permite pausar la corutina y ceder el control para que otras corutinas avancen.
2. **asyncio.run(main())**: Es la forma recomendada para ejecutar el bucle de eventos asíncrono en el programa principal.
3. **Evita el bloqueo**: Las funciones sin await, como operaciones de CPU o funciones de espera síncronas, bloquearán el flujo de ejecución asíncrono.
### Ventajas del Asincronismo con asyncio
- **Rendimiento**: Permite manejar miles de operaciones I/O sin necesidad de crear múltiples hilos o procesos.
- **Escalabilidad**: Ideal para aplicaciones que necesitan manejar muchas conexiones simultáneamente, como servidores web.
- **Eficiencia**: Reduce el consumo de memoria y CPU al evitar el uso de múltiples procesos o hilos.
### Resumen
- **Asincronismo** con asyncio permite que las tareas de entrada y salida se ejecuten sin bloquear el flujo.
- **Corutinas** (async def) y **await** son esenciales para ejecutar código de forma asíncrona.
- **asyncio.gather** ejecuta varias tareas en paralelo.
- asyncio es ideal para tareas de I/O intensivas y es la solución recomendada para aplicaciones que necesitan alto rendimiento en operaciones concurrentes de red o archivo.
Este enfoque permite crear aplicaciones eficientes y escalables en entornos de alto rendimiento.
"""Implementar un sistema de descargade archivos asincronica, donde cada archivotarde un tiempo variable en descargarse
"""import randomimport timeimport asyncioimport json
file1={'nombre':'babyshower', 'hora':'12:22' ,'precio':3000}file2={'nombre':'birthday', 'hora':'11:22' ,'precio':40000}
async def download_file(file1,file2): print("Iniciando descarga") #simulamos la operacion tiempo=random.randint(4,9) await asyncio.sleep(tiempo) with open('file1.json',mode='w')as file: json.dump(file1,file,indent=4) print("Descarga de file1 procesada") await asyncio.sleep(tiempo) with open('file2.json',mode='w')as file: json.dump(file2,file,indent=4) print("Descarga de file2 procesada")
async def main(): print(f"Inicio de descarga") result=await download_file(file1,file2) print("Descarga finalizada")
asyncio.run(main())
Gracias por tu aporte
Les dejo mi solucion, cualquier feedback es bienvenido.
.
.
.
.
.
.
.
.
.
### RETO# Implementar un sistema de descarga de archivos asincrona, donde cada archivo tarde un tiempo variable en descargarse
import asyncio
import random
archivero_list =["archivo.txt","musica.mp3","video.mp4","youtube.yu"]async def descargar_archivos(archivero):for archivo inarchivero: random_time = random.randint(1,10)print(f"Descargando {archivo}")await asyncio.sleep(random_time)print(f"{archivo} descargado.")return"Todos los archivos han sido descargados con exito 🎉"async def main():print("Inicio de las descargas") result =awaitdescargar_archivos(archivero_list)print(result)asyncio.run(main())
import asyncio
async def proccess_data(data):"""
Esta función simula el procesamiento de datos de forma asíncrona.Args:data:Los datos a procesar(en este caso, un string).Returns:El resultado del procesamiento(en este caso, el string duplicado)."""
print("Procesando datos...") # Indica que el procesamiento ha comenzado.await asyncio.sleep(2) # Simula una operación que tarda 2segundos(por ejemplo, lectura de un archivo, acceso a una base de datos, etc.).Este es el punto clave donde se "cede" el control a otras tareas asíncronas.print("Datos procesados") # Indica que el procesamiento ha terminado.return data *2 # Duplica los datos(simplemente como ejemplo de transformación).Podría ser cualquier otra operación.async def main():"""
Función principal que coordina el procesamiento asíncrono."""
print("Inicio del procesamiento") # Indica el inicio del programa.data=awaitproccess_data('archivo.txt') # Llama a la función de procesamiento de forma asíncrona y espera su resultado.La palabra clave 'await' es crucial aquí; permite que otras tareas se ejecuten mientras 'proccess_data' está "esperando"(durante el sleep).print(f"Resultado: {data} ") # Imprime el resultado del procesamiento.asyncio.run(main()) # Ejecuta la función principal de forma asíncrona.`asyncio.run()` es la forma recomendada de iniciar un programa asíncrono en Python3.7+.Crea un bucle de eventos y ejecuta la corutina `main()`.# Explicación detallada del comportamiento asíncrono:# 1.`asyncio.run(main())`:Comienza la ejecución del programa asíncrono.Se crea un bucle de eventos.# 2.`main()`:Imprime"Inicio del procesamiento" y luego llama a `await proccess_data('archivo.txt')`.# 3.`proccess_data()`:Imprime"Procesando datos..." y luego llega a `await asyncio.sleep(2)`.Aquí es donde ocurre la magia del asincronismo.En lugar de bloquear el programa durante 2 segundos,`asyncio.sleep(2)` le dice al bucle de eventos:"Esta tarea (proccess_data) va a tardar un poco. Mientras tanto, puedes ejecutar otras tareas que estén listas para ejecutarse".# 4.El bucle de eventos:Mientras`proccess_data` está "durmiendo", el bucle de eventos podría ejecutar otras corutinas(si las hubiera).En este ejemplo, no hay otras, pero en un programa real, aquí se podrían manejar otras operaciones de E/S, como solicitudes de red, lectura de archivos, etc.,*sin bloquear el programa principal*.# 5.Después de 2segundos:`asyncio.sleep(2)` se completa.El bucle de eventos "despierta" a `proccess_data` y continúa su ejecución.# 6.`proccess_data()`:Imprime"Datos procesados" y devuelve el resultado.# 7.`main()`:Recibe el resultado, lo imprime y termina.# La clave del asincronismo es el uso de `await`.`await` solo se puede usar dentro de funciones `async`.Cuando se encuentra un `await`, la función cede el control al bucle de eventos, permitiendo que otras tareas se ejecuten.Esto permite que un programa realice múltiples operaciones de E/S concurrentemente sin necesidad de hilos(threads) o procesos separados, lo que simplifica el código y mejora el rendimiento en muchos casos.```import asyncio
async def proccess\_data(data): """ Esta función simula el procesamiento de datos de forma asíncrona.
Args: data: Los datos a procesar (en este caso, un string).
Returns: El resultado del procesamiento (en este caso, el string duplicado). """ print("Procesando datos...") *# Indica que el procesamiento ha comenzado.* await asyncio.sleep(2) *# Simula una operación que tarda 2 segundos (por ejemplo, lectura de un archivo, acceso a una base de datos, etc.). Este es el punto clave donde se "cede" el control a otras tareas asíncronas.* print("Datos procesados") *# Indica que el procesamiento ha terminado.*
return data \* 2 *# Duplica los datos (simplemente como ejemplo de transformación). Podría ser cualquier otra operación.*
async def main(): """ Función principal que coordina el procesamiento asíncrono. """ print("Inicio del procesamiento") *# Indica el inicio del programa.* data = await proccess\_data('archivo.txt') *# Llama a la función de procesamiento de forma asíncrona y espera su resultado. La palabra clave 'await' es crucial aquí; permite que otras tareas se ejecuten mientras 'proccess\_data' está "esperando" (durante el sleep).* print(f"Resultado: {data} ") *# Imprime el resultado del procesamiento.*
asyncio.run(main()) *# Ejecuta la función principal de forma asíncrona. `asyncio.run()` es la forma recomendada de iniciar un programa asíncrono en Python 3.7+. Crea un bucle de eventos y ejecuta la corutina `main()`.*
*# Explicación detallada del comportamiento asíncrono:*
*# 1. `asyncio.run(main())`: Comienza la ejecución del programa asíncrono. Se crea un bucle de eventos.*
*# 2. `main()`: Imprime "Inicio del procesamiento" y luego llama a `await proccess\_data('archivo.txt')`.*
*# 3. `proccess\_data()`: Imprime "Procesando datos..." y luego llega a `await asyncio.sleep(2)`. Aquí es donde ocurre la magia del asincronismo. En lugar de bloquear el programa durante 2 segundos, `asyncio.sleep(2)` le dice al bucle de eventos: "Esta tarea (proccess\_data) va a tardar un poco. Mientras tanto, puedes ejecutar otras tareas que estén listas para ejecutarse".*
*# 4. El bucle de eventos: Mientras `proccess\_data` está "durmiendo", el bucle de eventos podría ejecutar otras corutinas (si las hubiera). En este ejemplo, no hay otras, pero en un programa real, aquí se podrían manejar otras operaciones de E/S, como solicitudes de red, lectura de archivos, etc., \*sin bloquear el programa principal\*.*
*# 5. Después de 2 segundos: `asyncio.sleep(2)` se completa. El bucle de eventos "despierta" a `proccess\_data` y continúa su ejecución.*
*# 6. `proccess\_data()`: Imprime "Datos procesados" y devuelve el resultado.*
*# 7. `main()`: Recibe el resultado, lo imprime y termina.*
*# La clave del asincronismo es el uso de `await`. `await` solo se puede usar dentro de funciones `async`. Cuando se encuentra un `await`, la función cede el control al bucle de eventos, permitiendo que otras tareas se ejecuten.Esto permite que un programa realice múltiples operaciones de E/S concurrentemente sin necesidad de hilos(threads) o procesos separados, lo que simplifica el código y mejora el rendimiento en muchos casos.*
#Este es mi aporte import asyncio
import random
asyncdefdownload_process(data): jump ="\n"print(f'Downloading the file in the url: {jump}{data}') number = random.randint(1,10)await asyncio.sleep(number)print(f"File download done")return data
asyncdefprocess_data(data):withopen(data, mode='r')as archivo: reader = archivo.read()print(f'Processing {archivo}....')#Simulate operation await asyncio.sleep(10)print(f'{reader} processed data')return data
asyncdefmain(url_file):print('Start processing') result =await process_data(url_file)print(f'Processing finished {result}')await asyncio.sleep(4)print("------------------------------------------------------>")print('Start download') download =await download_process(url_file)print(f"Sucessfully download {download}")data ="./good_practices/processing_tasks/archivo.txt"asyncio.run(main(data))
import asyncio
import random
files =[{'name':'archivo1.txt'},{'name':'archivo2.txt'},{'name':'archivo3.txt'}]asyncdefdownload_file(data):forfilein data:print("Descargando archivo..")await asyncio.sleep(random.randint(5,10))print(f"{file.get('name')} descargado con exito")asyncio.run(download_file(files))
import asyncio
import random
async def process_data(data: str)-> str:"""Funciónasíncrona que simula un procesamiento de datos
Parametros:data(str):Datos a procesar"""
print(f"Procesando {data}...") #Simular una opereción
await asyncio.sleep(random.randint(1,10))print(f"Procesamiento de {data} completado")return data.upper()async def main():print("Inicio de procesamiento") result1 =awaitprocess_data("archivo.txt") result2 =awaitprocess_data("archivo2.txt") result3 =awaitprocess_data("archivo3.txt")print(f"""
Resultado1:{result1}Resultado2:{result2}Resultado3:{result3}""")asyncio.run(main())
import asyncio, random
asyncdefprocess_data(data, sleep:int):print(f'Procesando {data}...')#Simular una operaciónawait asyncio.sleep(sleep)print(f'{data} procesado.')return data *2asyncdefmain():print('Inicio de procesamiento')for i inrange(10): result =await process_data(f"archivo_{i}.txt", sleep=random.randint(1,5))print(f'Resultado: {result}')asyncio.run(main())
import asyncio, randomasync def process_data(data, sleep: int): print(f'Procesando {data}...') #Simular una operación await asyncio.sleep(sleep) print(f'{data} procesado.') return data * 2async def main(): print('Inicio de procesamiento') for i in range(10): result = await process_data(f"archivo_{i}.txt", sleep=random.randint(1, 5)) print(f'Resultado: {result}')asyncio.run(main())
import asyncio
import random
import time
lista_archivos =['archivo1.txt','archivo2.txt','archivo3.txt']lista_resultados =[]asyncdefdescarga_archivo(archivo): tiempo = random.randint(1,10)print(f'Inicianado descarga archivo {archivo}...')await asyncio.sleep(tiempo)print(f'Descarga del archivo {archivo} terminada a {time.strftime('%X')}')return archivo
asyncdefmain():print(f'Inicio de procesamiento') dict_tareas ={}#Lanzamos tareasfor i inrange(len(lista_archivos)): dict_tareas[i]= asyncio.create_task(descarga_archivo(lista_archivos[i]))#resultado = await descarga_archivo(lista_archivos[i])#Esperando que todas acabenfor i inrange(len(lista_archivos)):await dict_tareas[i]print(f'Tereas finalizadas')asyncio.run(main())
No entendí muy bien, la verdad, me guié con Gemini y me quedó un poco más claro, acá mi código:
import asyncio
import random
asyncdefdescargar_archivo(nombre): tiempo = random.randint(1,5)print(f'Iniciando descarga {nombre} tardará {tiempo}s...')await asyncio.sleep(tiempo)print(f'Archivo {nombre} descargado en {tiempo}s')returnf'{nombre} ok'asyncdefmain(): archivos =['base_datos.csv','modelo_ia.bin','ventas_2025.xlsx','logo.png'] tareas =[descargar_archivo(f)for f in archivos]print('Iniciando descagas masivas') resultados =await asyncio.gather(*tareas)print('\n---Reporte Final---')print(f'Descargas completadas: {resultados}')if __name__ =='__main__': asyncio.run(main())
Me preguntó porque no hizo el ejemplo haciendo una petición real a una api
import random
import asyncio
async def download_documents(document):print(f'Downloading {document}...')await asyncio.sleep(random.randrange(1,10))print(f'{document} has been downloaded.')returnTrueasync def manager(documents: list):print('Start processing.')for file indocuments: result =awaitdownload_documents(file)print(f'Document {file} is ready: {result}.')docs =['text.txt','file.zip','check.pdf']asyncio.run(manager(docs))
import asyncio
importrandomfrom datetime import datetime
async def download_file(filename: str,min_time: float =1.0,max_time: float =5.0):"""
Simula la descarga de un archivo con un tiempo variable.Args:filename:Nombre del archivo a descargar
min_time:Tiempo mínimo de descarga en segundos(default:1.0)max_time:Tiempo máximo de descarga en segundos(default:5.0)Returns:str:Mensaje de confirmación con el tiempo de descarga
"""
# Generar un tiempo aleatorio de descarga
download_time = random.uniform(min_time, max_time) start_time = datetime.now()print(f'[{start_time.strftime("%H:%M:%S")}] Iniciando descarga de {filename}...')print(f' Tiempo estimado: {download_time:.2f} segundos') # Simular la descarga esperando el tiempo aleatorio
await asyncio.sleep(download_time) end_time = datetime.now() elapsed_time =(end_time - start_time).total_seconds()print(f'[{end_time.strftime("%H:%M:%S")}] ✓ {filename} descargado exitosamente ({elapsed_time:.2f}s)')return f'{filename} completado en {elapsed_time:.2f}s'async def main():"""
Función principal que ejecuta múltiples descargas en paralelo."""
# Lista de archivos a descargar
files =['documento1.pdf','imagen.jpg','video.mp4','datos.csv','presentacion.pptx']print('='*60)print('Sistema de Descarga Asincrónica')print('='*60)print(f'Total de archivos: {len(files)}\n') # Ejecutar todas las descargas en paralelo
start_total = datetime.now() results =await asyncio.gather(*[download_file(file)for file in files]) end_total = datetime.now() total_time =(end_total - start_total).total_seconds()print('\n'+'='*60)print('Resumen de descargas:')print('='*60)for result inresults:print(f' • {result}')print(f'\nTiempo total de ejecución: {total_time:.2f} segundos')print(f'Tiempo que habría tomado secuencialmente: ~{sum([random.uniform(1.0, 5.0) for _ in files]):.2f} segundos')if __name__ =='__main__': asyncio.run(main())
"""
Implementar un sistema de descarga de archivos
asíncrona, donde cada archivo tarde un tiempo variable
en descargarse.
"""import asyncio
import random
asyncdefdownload_file(file_name:str):# Se simulará el tiempo de descarga con el random tiempo = random.randint(1,5)print(f"Descargando {file_name}...| Tiempo estimado: {tiempo} segundos")try:await asyncio.sleep(tiempo)except Exception as e:print(f"Error al descargar {file_name}: {e}")print(f"Descarga completada para {file_name}")asyncdefmain(): files =["LOL.exe","HoyoPlay.exe","Varesa.png","Build Legendaria.mp4","Nod-Krai.mp3",] tasks =[download_file(file)forfilein files]print("Iniciando descargas...")await asyncio.gather(*tasks)print(f"Todas las descargas fueron completadas {files}")if __name__ =="__main__": asyncio.run(main())
import asyncio
import random
import time
import sys
# --- CORRUTINA DE DESCARGA CON BARRA DE PROGRESO ---
# Aseguramos que el archivo de errores esté limpio al iniciar o lo creamos
with open("errores_descarga.txt", "a") as f: pass
asyncio.run(sistema_principal())
import asyncio
import random
async def descargar_archivo(nombre_archivo): tiempo_descarga = random.randint(1,5)print(f"Descargando {nombre_archivo}...")await asyncio.sleep(tiempo_descarga)print(f"{nombre_archivo} descargado en {tiempo_descarga} segundos.")async def main(): archivos =[f"archivo_{i}.txt"for i inrange(5)]await asyncio.gather(*(descargar_archivo(archivo)for archivo in archivos))asyncio.run(main())
mi solución al reto:
'''
Crear un sistema de gestión de descargas. Este sistema debe asumir que cada descarga tiene un tiempo de procesamiento diferente, y el tiempo
debe ser determinado de manera aleatoria.
'''import asyncio
import random
files:dict={"diary":"diary.txt","book":"book.txt","curriculum":"curriculum.pdf","photo":"photo.jpg"}asyncdefdownload_files(files:dict)->list: downloaded_files:list=[]for key, value in files.items():#recorremos las llaves y los valores de files random_sleep = random.randint(1,7)print(f'Starting the download process of the {key} file (estimated time: {random_sleep} seconds)...')await asyncio.sleep(random_sleep)print(f'Download completed! file: {value}') downloaded_files.append(value)return downloaded_files #retornamos una lista con los archivos descargadosasyncdefmain():print('--Starting main--\n') result:list=await download_files(files)print(f'\n--Process completed--\nOutput: {result}')asyncio.run(main())#corremos el proceso principal
