Un classico esercizio che anche qui sul sito abbiamo visto in molti linguaggi e tecnologie. Questa volta lo affrontiamo in python utilizzando socket e dinamica client/server. Non teniamo conto di multi client con concorrenza ma vogliamo realizzare un client capace di inviare ad un server una sorta di lista delle cose da fare specificando un testo/descrizione, una tipologia prefissata ed una durata numerica in ore. Il server è capace di archiviare la lista ed eseguire le operazioni di cancellazione, visualizzazione, individuare il task con durata maggiore, minore o una media dei valori archiviati.
Indice dei contenuti
La classe Task
Gli esercizi precedenti, come Server/Client prenotazione in python o Server/Client Ping Pong in Python erano piuttosto semplici. Al di la dello scambio di messaggi client/server i dati e comandi inviati erano semplici e le informazioni archiviate riconducibili a dati primitivi (un singolo testo, piuttosto che un valore intero per i posti). Qui facciamo un passetto in più perché per ogni dato gestito dal server, abbiamo bisogno di un contenitore più strutturato come una classe. Proponiamo quindo la classe Task che conterrà descrizione. tipologia, durata e un timestamp per annotare quando tale task è stato creato. Per semplicità abbiamo scritto i metodi set e un metodo per tornare una stringa completa di tutti i gli attributi di un oggetto Task.
from Tipologia import Tipologia
import time
class Task:
def __init__(self, _descrizione: str = '', _tipologia: Tipologia = Tipologia.CASA, _durata: int = 0):
self.descrizione = _descrizione
self.tipologia = _tipologia
self.durata = _durata
self.timestamp = time.time()
def setDescrizione(self, _descrizione: str) ->None:
self.descrizione = _descrizione
def setTipologia(self, _tipologia: Tipologia) ->None:
self.descrizione = _tipologia
def setDurata(self, _durata: int) ->None:
self.durata = _durata
def toString(self) ->str:
return self.descrizione + " " + self.tipologia + " " + str(self.durata) + " " + str(self.timestamp)
La tipologia poi la possiamo gestire come stringa semplice, oppure utilizzare un classico enum che in realtà in python si scrivi comunque con una particolare classe
from enum import Enum
class Tipologia(Enum):
CASA = 1
LAVORO = 2
HOBBY = 3Il server
A questo punto siamo pronti a scrivere il nostro server. Tra i suoi attributi spicca il dizionario che conterrà un indice numerico progressivo e il relativo oggetto Task.
Tra i metodi, definiamo un metodo toString() che altro non fa che restituire una stringa composta dall’ id numerico progressivo dell’elenco più le informazioni contenute all’interno del Task stesso
def toString(self) -> str:
stampa = ''
for id,task in self.elenco.items():
stampa += str(id) +" " +task.toString()
return stampaA differenza degli esercizi citati, per capire quale comando viene inviato dal client facciamo un ulteriore variazione possibile. Prelevo una sottostriga da sinistra di 4 caratteri, lunghezza massima del comando LIST e QUIT, mentre gli altri ADD, DEL, MIN MAX, AVG sono tutti lunghi tre caratteri. Se la sosttostringa presenta quindi uno spazio, si tratterà sicuramente di un comando da tre lettere e mi basta eliminare lo spazio vuoto.
comando = messaggio[:4].strip() #prendo i primi 4 caratteri e al limite tolgo lo spazio vuoto se sono 3Poi quindi procedo con il solito costrutto match per eseguire le richieste di routine e rispondere al client. Non commentiamo più di tanto le operazioni per individuare min, max e media avg perché consistono banalmente nei meccanismi base di scorrere l’intero dizionario, estrapolare il task e la sua durata per poter individuare o sommare per la richiesta.
import socket
from Task import Task
class ServerTask:
def __init__(self, _host="127.0.0.1", _porta =5000):
self.host = _host
self.porta = _porta
self.servizio = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
self.id = 0 #id progressivo dei task
self.elenco = {} #dizionario dove aggiungere i task
def toString(self) -> str:
stampa = ''
for id,task in self.elenco.items():
stampa += str(id) +" " +task.toString()
return stampa
def start(self) -> None:
self.servizio.bind((self.host,self.porta))
self.servizio.listen(1)
print("Servizio attivo su ", self.host, self.porta)
while True:
connessione, indirizzo = self.servizio.accept()
print("Connessione ricevuta da ", indirizzo)
messaggio = connessione.recv(1024).decode()
comando = messaggio[:4].strip() #prendo i primi 4 caratteri e al limite tolgo lo spazio vuoto se sono 3
match comando:
case "ADD":
comando, testo, tipologia, giorni = messaggio.split(" ",4)
self.id += 1
task = Task(testo,tipologia, int(giorni))
self.elenco[self.id] = task #aggiungo un nuovo task
risposta = "OK:" + str(self.id) + " ts="+ str(task.timestamp)
connessione.send(risposta.encode())
case "DEL":
comando, id = messaggio.split(" ",2)
if self.elenco[id]:
del self.elenco[id]
connessione.send("OK".encode())
else:
connessione.send("ERR non trovato".encode())
case "LIST":
lista = self.toString()
connessione.send(lista.encode())
case "MIN":
giorno_min = -1 #o metti un numero enorme o forzi il primo giro
for id,task in self.elenco.items():
if task.durata < giorno_min or giorno_min == -1:
giorno_min = task.durata
connessione.send(str(giorno_min).encode())
case "MAX":
giorno_max= 0
for id,task in self.elenco.items():
if task.durata > giorno_max:
giorno_max = task.durata
connessione.send(str(giorno_max).encode())
case "AVG":
somma_giorni= 0
for id,task in self.elenco.items():
somma_giorni += task.durata
avg = somma_giorni / len(self.elenco)
connessione.send(str(avg).encode())
case "QUIT":
print("Connessione in chiusura...")
connessione.close()
exit()
case "_": # restituisce un errore
pass
if __name__ == "__main__":
server = ServerTask()
server.start()Ovviamenete è fortemente migliorabile, valutando opportunamente la correttezza dei dati inseriti, dell’enum.
Il client
In realtà il client assomiglia a tutti gli esercizi che abbiamo visto sulla gestione dei socket python. Si limita a chiedere all’utente una lista comando ed inviarla al server. Non facciamo particolari controlli di corretta e valida immissione del comando e dei suoi parametri. Tale funzionalità per il momento non la implementiamo se non in parte sul server.
import socket
class ClientTask:
def __init__(self, _host = "127.0.0.1", _porta = 5000):
self.host = _host
self.porta = _porta
self.servizio = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
def start(self) -> None:
print("Comandi ADD/DEL/LIST/MIN/MAX/AVG/QUIT")
print("es. ADD pippo CASA 1, DEL 4, LIST, QUIT")
comando = ""
while True and comando != "QUIT":
self.servizio = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.servizio.connect((self.host,self.porta))
comando = input(">")
self.servizio.send(comando.encode())
risposta = self.servizio.recv(1024).decode()
print(risposta)
self.servizio.close()
print("Connesione chiusa")
if __name__ == "__main__":
client = ClientTask()
client.start()Ultima modifica 30 Ottobre 2025
