Esercizi Asyncio Python

Codegrind Team•Jul 10 2024

Scopri la programmazione asincrona con asyncio in Python attraverso questi esercizi introduttivi.

Esercizio 1

Creare un'attività asincrona che stampa i numeri da 1 a 10 con un intervallo di un secondo tra ciascuno.

import asyncio

async def print_numbers():
    for i in range(1, 11):
        print(i)
        await asyncio.sleep(1)

async def main():
    await print_numbers()

asyncio.run(main())

Esercizio 2

Scrivere un programma che avvia due task asincroni; uno stampa numeri pari e l'altro numeri dispari fino a 10.

import asyncio

async def print_even():
    for i in range(2, 11, 2):
        print(f"Pari: {i}")
        await asyncio.sleep(1)

async def print_odd():
    for i in range(1, 11, 2):
        print(f"Dispari: {i}")
        await asyncio.sleep(1)

async def main():
    task1 = asyncio.create_task(print_even())
    task2 = asyncio.create_task(print_odd())
    await task1
    await task2

asyncio.run(main())

Esercizio 3

Implementare un task asincrono che esegue un countdown da 10 a 1 e stampa "Fine!" al termine.

import asyncio

async def countdown():
    for i in range(10, 0, -1):
        print(i)
        await asyncio.sleep(1)
    print("Fine!")

async def main():
    await countdown()

asyncio.run(main())

Esercizio 4

Creare un task asincrono che calcola la somma di un range di numeri e poi stampa il risultato.

import asyncio

async def calculate_sum():
    total = sum(range(1, 1001))
    await asyncio.sleep(2)  # Simulazione di un'operazione lunga
    print(f"La somma è: {total}")

async def main():
    await calculate_sum()

asyncio.run(main())

Esercizio 5

Utilizzare asyncio per eseguire una funzione che simula il download di file da diverse URL in modo asincrono.

import asyncio
import random

async def download_file(url):
    print(f"Inizio download da {url}")
    await asyncio.sleep(random.randint(1, 3))  # Simula un tempo di download variabile
    print(f"Download completato da {url}")

async def main():
    urls = ['http://example.com/a', 'http://example.com/b', 'http://example.com/c']
    tasks = [asyncio.create_task(download_file(url)) for url in urls]
    await asyncio.gather(*tasks)

asyncio.run(main())

Esercizio 6

Scrivere un programma con asyncio che gestisce un timer che scatta ogni 5 secondi, stampando "Timer scattato!".

import asyncio

async def timer():
    while True:
        await asyncio.sleep(5)
        print("Timer scattato!")

async def main():
    await timer()

asyncio.run(main())

Esercizio 7

Creare una funzione asincrona che attende l'input dell'utente e poi stampa quello che è stato inserito.

import asyncio

async def get_input():
    result = await asyncio.to_thread(input, "Inserisci qualcosa: ")
    print(f"Hai inserito: {result}")

async def main():
    await get_input()

asyncio.run(main())

Esercizio 8

Implementare una funzione asincrona che esegue un task che può essere cancellato esternamente dopo un certo tempo.

import asyncio

async def cancellable_task():
    try:
        while True:
            print("Esecuzione...")
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Task cancellato")

async def main():
    task = asyncio.create_task(cancellable_task())
    await asyncio.sleep(5)  # Permette al task di eseguire per un po'
    task.cancel()
    await task

asyncio.run(main())

Esercizio 9

Usare asyncio per implementare un semplice echo server che ascolta su una porta TCP.

import asyncio

async def handle_client(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print(f"Ricevuto {message} da {addr}")
    writer.write(data)
    await writer.drain()
    writer.close()

async def main():
    server = await asyncio.start_server(handle_client, '127.0.0.1', 8888)
    addr = server.sockets[0].getsockname()
    print(f'Servendo su {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

Esercizio 10

Creare un programma asincrono che attende la terminazione di più task asincroni, usando `asyncio.wait` con un timeout.

import asyncio

async def task(name, time_to_sleep):
    await asyncio.sleep(time_to_sleep)
    print(f"Task {name} completato")

async def main():
    tasks = [task(f"Task {i}", i) for i in range(1, 4)]
    done, pending = await asyncio.wait(tasks, timeout=2)
    for task in done:
        print(f"{task.get_name()} finito")
    for task in pending:
        print(f"{task.get_name()} non finito e cancellato")
        task.cancel()

asyncio.run(main())