Fundamentos de Programación y Python

1

¿Por qué aprender Python?

2

Introducción a Python

3

Conceptos Básicos de Programación

4

Práctica: Te doy la bienvenida a los ejercicios interactivos

5

Manipulación de Cadenas de Texto en Python

6

Enteros, Flotantes y Booleanos

7

Todo lo que Debes Saber sobre print en Python

8

Operaciones Matemáticas en Python

9

Operaciones de Entrada/Salida en Consola

Colección y Procesamiento de Datos en Python

10

Listas

11

Método slice

12

Listas de más dimensiones y Tuplas

13

Aplicación de Matrices

14

Diccionarios

15

Comprehension Lists en Python (CLASE NUEVA)

Control de Flujo en Python

16

Estructuras condicionales

17

Bucles y Control de Iteraciones

18

Generadores e Iteradores

Funciones y Manejo de Excepciones en Python

19

Uso de Funciones en Python

20

Funciones Lambda y Programación Funcional en Python

21

¿Cómo realizar una función recursiva en Python?

22

Manejo de Excepciones y Uso de Pass (CLASE NUEVA)

Programación Orientada a Objetos en Python

23

Fundamentos de Programación Orientada a Objetos en Python

24

Ejercicio Biblioteca con POO

25

Herencia en POO con Python

26

Objetos heredados

27

Los 4 pilares de la programacion orientada a objetos

28

Uso de super() en Python (CLASE NUEVA)

29

Superando los Fundamentos de Programación Orientada a Objetos en Python

Lectura y escritura de archivos

30

Manejo de Archivos .TXT (CLASE NUEVA)

31

Manejo de Archivos CSV (CLASE NUEVA)

32

Manejo de Archivos JSON (CLASE NUEVA)

Biblioteca estándar de Python

33

Biblioteca estándar en Python (CLASE NUEVA)

34

Librería Os, Math y Random (CLASE NUEVA)

35

Librería Statistics y Análisis Estadístico (CLASE NUEVA)

36

Proyecto final: Guerra naval

Conceptos avanzados de Python

37

Recapitulación de lo aprendido hasta ahora

38

Escribir código Pythonico y profesional

39

Comentarios y Docstrings en Python

40

Scope y closures: variables locales y globales

41

Anotaciones de tipo

42

Validación de tipos en métodos

43

Librería Collections y Enumeraciones

Decoradores

44

Decoradores en Python

45

Decoradores anidados y con parámetros

46

Uso de Decoradores en clases y métodos

Métodos y estructura de clases en Python

47

Métodos mágicos

48

Sobrecarga de operadores

49

Implementación de `if __name__ == "__main__":`

50

Metaprogramación en Python

51

Uso de *args y **kwargs

52

Métodos privados y protegidos

53

Gestión avanzada de propiedades

54

Métodos estáticos y de clase avanzados

Programación concurrente y asíncrona

55

Introducción a la concurrencia y paralelismo

56

Threading y multiprocessing en Python

57

Asincronismo con asyncio

58

Asincronismo y concurrencia

Creación de módulos y paquetes

59

Creación de módulos en Python

60

Gestión de paquetes

61

Publicación de paquetes en PyPI

Proyecto final

62

Implementación de un sistema completo

63

Implementación de un Sistema Completo

No tienes acceso a esta clase

¡Continúa aprendiendo! Únete y comienza a potenciar tu carrera

Curso de Python

Curso de Python

Carli Code

Carli Code

Asincronismo con asyncio

57/63
Recursos

Aportes 32

Preguntas 1

Ordenar por:

¿Quieres ver más aportes, preguntas y respuestas de la comunidad?

```js 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=await donwload_process('file1.txt') result1=await donwload_process('file2.txt') result2=await donwload_process('file3.txt') print(f'{result,result1,result2}') asyncio.run(main()) ```
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. ```python import asyncio async def tarea(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 async def main(): 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`. ```python import asyncio async def 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}" async def main(): 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.
```python #Este es mi aporte import asyncio import random async def download_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 async def process_data(data): with open(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 async def main(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)) ```
```python import asyncio import random files = [ {'name' : 'archivo1.txt'}, {'name' : 'archivo2.txt'}, {'name' : 'archivo3.txt'} ] async def download_file(data): for file in data: print("Descargando archivo..") await asyncio.sleep(random.randint(5,10)) print(f"{file.get('name')} descargado con exito") asyncio.run(download_file(files)) ```
![](https://static.platzi.com/media/user_upload/imagen-23728cd5-663f-473e-8f11-a441e2400797.jpg)
```python import asyncio import time async def descargar_imagen(tiempo, imagen): await asyncio.sleep(tiempo) print(imagen) async def main(): print(f"Iniciando descarga {time.strftime('%X')}") await descargar_imagen(3, 'Imagen_logo_python.png') await descargar_imagen(2, 'Imagen_logo_PHP.png') print(f"archivos descargados {time.strftime('%X')}") asyncio.run(main()) ```
```js import asyncio import random async def process_data(data: str) -> str: """Función así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 = await process_data("archivo.txt") result2 = await process_data("archivo2.txt") result3 = await process_data("archivo3.txt") print(f""" Resultado1: {result1} Resultado2: {result2} Resultado3: {result3} """) asyncio.run(main()) ```
```python import asyncio, random async def process_data(data, sleep: int): print(f'Procesando {data}...') #Simular una operación await asyncio.sleep(sleep) print(f'{data} procesado.') return data * 2 async 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, random` `async def process_data(data, sleep: int): print(f'Procesando {data}...') #Simular una operación await asyncio.sleep(sleep) print(f'{data} procesado.') return data * 2` `async 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())`
Para realmente sacarle partido a la asincronía hay que utilizar `asyncio.create_task` para que todas las tareas inicien a la vez y `asyncio.gather()` que esperarán a que todas terminen. En el siguiente código podéis comprobar la mejora que esto supondrá: ```python import asyncio import random import time async def download_process(file): print(f'Descargando archivo {file}') download_time = random.randint(1,10) await asyncio.sleep(download_time) print(f'{file} downloaded en {download_time} segundos') return file async def main(): start_time = time.time() print('Inicio de la descarga') # Versión secuencial (no vamos a obtener ninguna ventaja) print("\nMétodo secuencial:") result = await download_process('file1.txt') result1 = await download_process('file2.txt') result2 = await download_process('file3.txt') sequential_time = time.time() - start_time # Reiniciar el tiempo para la versión concurrente start_time = time.time() # Versión concurrente (mejorada) print("\nMétodo concurrente:") tasks = [ asyncio.create_task(download_process('file1.txt')), asyncio.create_task(download_process('file2.txt')), asyncio.create_task(download_process('file3.txt')) ] results = await asyncio.gather(*tasks) concurrent_time = time.time() - start_time print(f'\nTiempo secuencial: {sequential_time:.2f} segundos') print(f'Tiempo concurrente: {concurrent_time:.2f} segundos') print(f'Mejora de velocidad: {sequential_time/concurrent_time:.2f}x') asyncio.run(main()) ```
Para realmente sacarle partido a la asincronía hay que utilizar `asyncio.create_task` para que todas las tareas inicien a la vez y `asyncio.gather()` que esperarán a que todas terminen. En el siguiente código podéis comprobar la mejora que esto supondrá: import asyncioimport randomimport time async def download\_process(file):    print(f'Descargando archivo {file}')    download\_time = random.randint(1,10)    await asyncio.sleep(download\_time)    print(f'{file} downloaded en {download\_time} segundos')    return file async def main():    start\_time = time.time()     print('Inicio de la descarga')     # Versión secuencial (no vamos a obtener ninguna ventaja)    print("\nMétodo secuencial:")    result = await download\_process('file1.txt')    result1 = await download\_process('file2.txt')    result2 = await download\_process('file3.txt')    sequential\_time = time.time() - start\_time     # Reiniciar el tiempo para la versión concurrente    start\_time = time.time()     # Versión concurrente (mejorada)    print("\nMétodo concurrente:")    tasks = \[        asyncio.create\_task(download\_process('file1.txt')),        asyncio.create\_task(download\_process('file2.txt')),        asyncio.create\_task(download\_process('file3.txt'))    ]    results = await asyncio.gather(\*tasks)    concurrent\_time = time.time() - start\_time     print(f'\nTiempo secuencial: {sequential\_time:.2f} segundos')    print(f'Tiempo concurrente: {concurrent\_time:.2f} segundos')    print(f'Mejora de velocidad: {sequential\_time/concurrent\_time:.2f}x') asyncio.run(main())
Comparto un ejercicio diferente que realice. ![](https://static.platzi.com/media/user_upload/image-9feab36b-764c-4989-977d-71346447e322.jpg)
```python """ impletar un sistema de descarga de archivos asincrona, donde cada archivo tarde un tiempo variable en descargarse """ import asyncio import time from random import randint async def download_file(name:str, delay:int): await asyncio.sleep(delay) async_time = randint(1, 10) print(f"Se inicio la descarga del archivo {name}\n tiempo de espera de {async_time} segundos") await asyncio.sleep(async_time) print(f"Descarga finalizada {name}") files = [ { "name": 'Pokemon_rubi_omega.3ds', "time_delay": 2 }, { "name": 'Valorant.exe', "time_delay": 3 }, { "name": 'Genshin_impact.exe', "time_delay": 4 }, { "name": 'Mario_bros.gba', "time_delay": 1 } ] async def main(): for file in files: await download_file(file["name"], file["time_delay"]) asyncio.run(main()) ```
```python """ impletar un sistema de descarga de archivos asincrona, donde cada archivo tarde un tiempo variable en descargarse """ from multiprocessing import Pool from time import sleep from random import randint def download_file(name:str): time = randint(1, 10) print(f"Se inicio la descarga del archivo {name}\n tiempo de espera de {time} segundos") sleep(time) print(f"Descarga finalizada {name}") if __name__ == '__main__': files = ['Pokemon_rubi_omega.3ds', 'Valorant.exe', 'Genshin_impact.exe', 'Mario_bros.gba'] with Pool() as pool: pool.map(download_file, files) ```""" impletar un sistema de descarga de archivos asincrona,donde cada archivo tarde un tiempo variable en descargarse""" from multiprocessing import Poolfrom time import sleepfrom random import randint def download\_file(*name*:str):    time = randint(1, 10)    print(f"Se inicio la descarga del archivo {*name*}\n tiempo de espera de {time} segundos")    sleep(time)    print(f"Descarga finalizada {*name*}") if \_\_name\_\_ == '\_\_main\_\_':        files = \['Pokemon\_rubi\_omega.3ds', 'Valorant.exe', 'Genshin\_impact.exe', 'Mario\_bros.gba']     with Pool() as pool:        pool.map(download\_file, files)       
```python import asyncio import random async def donwload_process(file): print(f'Descargando archivo {file}') time = random.randint(1,15) await asyncio.sleep(time) print(f'{file} downloaded') return file async def main(): print('Inicio de la descarga') result=await donwload_process('file1.txt') result1=await donwload_process('file2.txt') result2=await donwload_process('file3.txt') print(f'{result,result1,result2}') asyncio.run(main()) ```
Mi solución: ```js import asyncio import random async def donwload_process(file): print(f'Descargadon archivo {file}...') await asyncio.sleep(random.randint(1,10)) print(f'{file} descargado') return file results = [] async def main(): for i in range(5): print("Inicio de la descarga!") results.append(await donwload_process(f'file{i}.txt')) print("Archivos descargados: ", *results, sep="\n") asyncio.run(main()) ```
![](https://static.platzi.com/media/user_upload/Captura%20de%20pantalla%202025-01-12%20164010-aa02cb04-f731-4ad1-9a68-c10e374f7b25.jpg)
```js import asyncio import random async def process_data(data): print(f'Procesando canciondes de {data}...') #Simular una operación random_number = random.randint(1,10) await asyncio.sleep(random_number) print(f'cancion de {data} numero {random_number}. procesado.') return "cancion descargada" async def main(): print('Inicio de procesamiento de descarga') result = await process_data('BARAK ') print(f'Resultado: {result}') asyncio.run(main()) asyncio.run(main()) asyncio.run(main()) asyncio.run(main()) ```import asyncioimport random async def process\_data(data): print(f'Procesando canciondes de {data}...') *#Simular una operación* random\_number = random.randint(1,10) await asyncio.sleep(random\_number) print(f'cancion de {data} numero {random\_number}. procesado.') return "cancion descargada" async def main(): print('Inicio de procesamiento de descarga') result = await process\_data('BARAK ') print(f'Resultado: {result}') asyncio.run(main())asyncio.run(main())asyncio.run(main())asyncio.run(main())
```js """ Implementar un sistema de descarga de archivos asincrónica, donce cada archivo tarde un tiempo variable en descargarse. """ import asyncio import random as rn async def download_file(file): print(f"Downloading {file}") await asyncio.sleep(rn.randint(5,10)) print(f"Downloaded {file}") async def main(): await asyncio.gather( download_file("file1"), download_file("file2"), download_file("file3") ) asyncio.run(main()) ```
```js ''' Implementar un sistema de descarga de archivos asincrona, donce cada archivo tarde un tiempo variable en descargable. ''' import asyncio import random as rn files = [ {'nombre': 'archivo1.txt', 'size': 10}, {'nombre': 'archivo2.txt', 'size': 20}, {'nombre': 'archivo3.txt', 'size': 30} ] async def download_file(data: dict) -> str: print(f"Process {data}...") await asyncio.sleep(rn.randint(10,60)) print(f"Document {data['nombre']} processed.") return f"Download: {data['nombre']}, tamaño: {data['size']}" async def main(): print('Download packeges asyncronus') tasks = [download_file(file) for file in files] results = await asyncio.gather(*tasks) print("Finish:", results) asyncio.run(main()) ```
```python import asyncio import random async def descargar_archivo(nombre, tiempo): print(f'Descargando: {nombre}, tiempo estimado: {tiempo} segundos...') await asyncio.sleep(tiempo) # Simula el tiempo de descarga print(f'Descarga completada de: {nombre}.') return nombre async def gestionar_descargas(lista_archivos): print("Iniciando el Gestor de Descargas...") # Crear la lista de tareas tareas = [ descargar_archivo(archivo, random.randint(1, 3)) # Llamamos a la funcion de descargar cada archivo, un tiempo aleatorio entre 1 y 3 segundos for archivo in lista_archivos ] resultados = await asyncio.gather(*tareas) # Ejecutar descargas en paralelo con gather print("Todas los archivos has sido descargados.") return resultados async def main(): archivos = ['archivo1.txt', 'imagen.jpg', 'archivo2.txt', 'comprimido.zip', 'archivo3.txt', 'documento.pdf', 'musica.mp3', 'archivo4.txt'] resultados = await gestionar_descargas(archivos) print(f'Archivos descargados: {resultados}') # Ejecutar el programa asyncio.run(main()) ```
Aquí está mi código para la resolución del reto planteado en esta clase:import asyncio, random \#funcion que simula la descarga de un archivoasync def process\_download(file\_name: str): download\_time = random.randint(5,15) print(f"Iniciando descarga del archivo {file\_name} (Tiempo estimado: {download\_time} segundos)...") await asyncio.sleep(download\_time) #simular operación print(f"Archivo '{file\_name}' descargado en {download\_time} segundos.") return file\_name, download\_time \#funcion principalasync def main(archivos: list): print("Iniciando de descargas...") print("--------------------------------------------------------------") tasks = \[asyncio.create\_task(process\_download(archivo)) for archivo in archivos] return await asyncio.gather(\*tasks), print(f"--------------------------------------------------------------\nTodas las descargas completadas") archivos = \["archivo1.rar", "archivo2.rar", "archivo3.rar"]asyncio.run(main(archivos)) ```js import asyncio, random #funcion que simula la descarga de un archivo async def process_download(file_name: str): download_time = random.randint(5,15) print(f"Iniciando descarga del archivo {file_name} (Tiempo estimado: {download_time} segundos)...") await asyncio.sleep(download_time) #simular operación print(f"Archivo '{file_name}' descargado en {download_time} segundos.") return file_name, download_time #funcion principal async def main(archivos: list): print("Iniciando de descargas...") print("--------------------------------------------------------------") tasks = [asyncio.create_task(process_download(archivo)) for archivo in archivos] return await asyncio.gather(*tasks), print(f"--------------------------------------------------------------\nTodas las descargas completadas") archivos = ["archivo1.rar", "archivo2.rar", "archivo3.rar"] asyncio.run(main(archivos)) ```
```js #descarga de archivos import random, asyncio archivos = ["archivo1.txt", "archivo2.txt", "archivo3.txt"] async def download_file(): archivo = int(input("selecciona un archivo [0,1,2]: ")) download_time = random.randint(1, 5) for i in range(len(archivos)): # Iterate through 0, 1, 2 if archivo == i: print(f"descargando {archivos[archivo]}...") await asyncio.sleep(download_time) print(f"archivo {archivos[archivo]} descargado con exito.") print(f"el tiempo de descarga fue: {download_time}seg") return archivos[archivo] async def main(): print("iniciando programa") results = await download_file() print(f"Archivo descargado: {results}") await main() ```
```js import random, asyncio archivos = ["archivo1.txt", "archivo2.txt", "archivo3.txt"] async def download_file(): archivo = int(input("selecciona un archivo: [0,1,2]")) download_time = random.randint(1,5) if archivo == 0: print(f"descargando {archivos[archivo]}") await asyncio.sleep(download_time) print(f"archivo {archivos[archivo]} descargado con exito.") print(f"el tiempo de descarga fue: {download_time}seg") return archivos[archivo] elif archivo == 1: print(f"descargando {archivos[archivo]}...") await asyncio.sleep(download_time) print(f"archivo {archivos[archivo]} descargado con exito.") print(f"el tiempo de descarga fue: {download_time}seg") return archivos[archivo] elif archivo == 2: print(f"descargando {archivos[archivo]}...") await asyncio.sleep(download_time) print(f"archivo {archivos[archivo]} descargado con exito.") print(f"el tiempo de descarga fue: {download_time}seg") return archivos[archivo] async def main(): print("iniciando programa") results = await download_file() print(f"Archivo descargado: {results}") await main() ```
Resolución de la actividad ```python import asyncio, random #funcion que simula la descarga de un archivo async def process_download(file_name: str): download_time = random.randint(5,20) #simula numeros entre el 5 y 15 aleatoriamente print(f"Iniciando descarga de archivo: {file_name}... Tiempo aproximado: {download_time} Segundos.") await asyncio.sleep(download_time) #simular operación print(f"Archivo '{file_name}' descargado en {download_time} segundos.") return file_name, download_time #funcion principal async def main(archivos: list): print("Inicio de descargas") tasks = [asyncio.create_task(process_download(archivo)) for archivo in archivos] return await asyncio.gather(*tasks), print(f"Todas las descargas completadas") archivos = ["Archivo_1.rar", "Archivo_2.rar", "Archivo_3.rar"] asyncio.run(main(archivos)) ```
`from random import randrangeimport asyncio` `async def download_file(file):    #Simular descarga    download_time = randrange(10)    print(download_time)    print(f"Descargando el archivo {file}...")    await asyncio.sleep(download_time)    return file` `async def main():    print(f"Se inicia el proceso de descarga...")    result = await download_file("Psicologia.pdf")    print(f"Se descargo correctamente el archivo {result}.")` `asyncio.run(main())`
```python import asyncio import random async def descargar_archivo(nombre, tiempo): print(f'Iniciando descarga de {nombre}, tiempo estimado: {tiempo} segundos...') await asyncio.sleep(tiempo) # Simula el tiempo de descarga print(f'Descarga completada de {nombre}.') return nombre async def gestionar_descargas(lista_archivos): print("Inicio del sistema de gestión de descargas...") tareas = [ descargar_archivo(archivo, random.randint(1, 5)) # Tiempo aleatorio entre 1 y 5 segundos for archivo in lista_archivos ] resultados = await asyncio.gather(*tareas) # Ejecutar descargas en paralelo print("Todas las descargas han finalizado.") return resultados async def main(): archivos = ['archivo1.txt', 'archivo2.txt', 'archivo3.txt', 'archivo4.txt'] resultados = await gestionar_descargas(archivos) print(f'Archivos descargados: {resultados}') # Ejecutar el programa asyncio.run(main()) ```
```python import asyncio import random async def download(data): print(f'Procesando {data}...') #Simular una operacion await asyncio.sleep(random.randint(1, 20)) print(f'{data} procesada.') return data async def main(): print('Inicio de procesamiento') result = await download('file.txt') print(f'Descarga finalizada, resultado: {result}') asyncio.run(main()) ```
```js import threading # Importamos el módulo threading para manejar hilos import random # Importamos random para generar tiempos aleatorios import time # Importamos time para simular pausas en el programa # Definimos la clase DownloadTask que hereda de threading.Thread class DownloadTask(threading.Thread): def __init__(self, file_name): super().__init__() # Inicializamos la clase base threading.Thread self.file_name = file_name # Guardamos el nombre del archivo a descargar # Método run que se ejecuta al iniciar el hilo con start() def run(self): # Generamos un tiempo de descarga aleatorio entre 1 y 5 segundos download_time = random.uniform(1, 5) # Mostramos un mensaje indicando el inicio de la descarga y el tiempo estimado print(f"Iniciando descarga de {self.file_name}, tiempo estimado: {download_time:.2f} segundos.") time.sleep(download_time) # Simulamos el tiempo de descarga con sleep() # Mostramos un mensaje indicando que la descarga ha finalizado print(f"Descarga completada de {self.file_name}.") # Definimos la clase DownloadManager para gestionar múltiples descargas class DownloadManager: def __init__(self): self.downloads = [] # Lista para almacenar las tareas de descarga activas # Método para agregar una nueva descarga al gestor def add_download(self, file_name): download_task = DownloadTask(file_name) # Creamos una nueva tarea de descarga self.downloads.append(download_task) # La agregamos a la lista de descargas download_task.start() # Iniciamos la tarea de descarga (hilo) # Método para esperar a que todas las descargas finalicen def wait_for_all(self): for task in self.downloads: task.join() # Esperamos a que cada hilo termine su ejecución # Punto de entrada del programa if __name__ == "__main__": manager = DownloadManager() # Creamos una instancia del gestor de descargas # Lista de archivos a descargar archivos = ['archivo1.txt', 'imagen2.jpg', 'documento3.pdf', 'musica4.mp3', 'comprimido5.zip'] # Iteramos sobre la lista de archivos y agregamos cada uno al gestor for archivo in archivos: manager.add_download(archivo) manager.wait_for_all() # Esperamos a que todas las descargas terminen print("Todas las descargas han finalizado.") # Mensaje final ```import threading # Importamos el módulo threading para manejar hilos import random # Importamos random para generar tiempos aleatorios import time # Importamos time para simular pausas en el programa \# Definimos la clase DownloadTask que hereda de threading.Thread class DownloadTask(threading.Thread): def \_\_init\_\_(self, file\_name): super().\_\_init\_\_() # Inicializamos la clase base threading.Thread self.file\_name = file\_name # Guardamos el nombre del archivo a descargar \# Método run que se ejecuta al iniciar el hilo con start() def run(self): \# Generamos un tiempo de descarga aleatorio entre 1 y 5 segundos download\_time = random.uniform(1, 5) \# Mostramos un mensaje indicando el inicio de la descarga y el tiempo estimado print(f"Iniciando descarga de {self.file\_name}, tiempo estimado: {download\_time:.2f} segundos.") time.sleep(download\_time) # Simulamos el tiempo de descarga con sleep() \# Mostramos un mensaje indicando que la descarga ha finalizado print(f"Descarga completada de {self.file\_name}.") \# Definimos la clase DownloadManager para gestionar múltiples descargas class DownloadManager: def \_\_init\_\_(self): self.downloads = \[] # Lista para almacenar las tareas de descarga activas \# Método para agregar una nueva descarga al gestor def add\_download(self, file\_name): download\_task = DownloadTask(file\_name) # Creamos una nueva tarea de descarga self.downloads.append(download\_task) # La agregamos a la lista de descargas download\_task.start() # Iniciamos la tarea de descarga (hilo) \# Método para esperar a que todas las descargas finalicen def wait\_for\_all(self): for task in self.downloads: task.join() # Esperamos a que cada hilo termine su ejecución \# Punto de entrada del programa if \_\_name\_\_ == "\_\_main\_\_": manager = DownloadManager() # Creamos una instancia del gestor de descargas \# Lista de archivos a descargar archivos = \['archivo1.txt', 'imagen2.jpg', 'documento3.pdf', 'musica4.mp3', 'comprimido5.zip'] \# Iteramos sobre la lista de archivos y agregamos cada uno al gestor for archivo in archivos: manager.add\_download(archivo) manager.wait\_for\_all() # Esperamos a que todas las descargas terminen print("Todas las descargas han finalizado.") # Mensaje final
1. **Definición de Corrutinas**: * `tarea_1` y `tarea_2` son funciones asíncronas definidas con `async def`. Utilizan `await asyncio.sleep()` para simular tareas que toman tiempo. 2. **Función Principal**: * `main` es otra función asíncrona que utiliza `asyncio.gather()` para ejecutar `tarea_1` y `tarea_2` de manera concurrente. * `asyncio.gather()` devuelve una lista de resultados una vez que todas las tareas han terminado. 3. **Ejecución del Bucle de Eventos**: * `asyncio.run(main())` se utiliza para ejecutar la función `main` y manejar el bucle de eventos.
Este ejemplo muestra cómo ejecutar múltiples tareas de manera asíncrona utilizando corrutinas: ```python import asyncio async def tarea_1(): print("Inicio de tarea 1") await asyncio.sleep(1) # Simula una tarea que toma 1 segundo print("Fin de tarea 1") return "Resultado de tarea 1" async def tarea_2(): print("Inicio de tarea 2") await asyncio.sleep(2) # Simula una tarea que toma 2 segundos print("Fin de tarea 2") return "Resultado de tarea 2" async def main(): # Ejecutar las tareas de manera concurrente resultados = await asyncio.gather(tarea_1(), tarea_2()) for resultado in resultados: print(resultado) # Ejecutar la función principal asyncio.run(main()) ```
### 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.