Hive.decorators

chunks(lst, n)

metodo per suddividere una lista madre in una nuova lista composta di sotto liste piu piccole

Parameters:

Name	Type	Description	Default
lst	list	lista da suddividere	required
n	int	numero di componenti dentro ogni sotto suddivisione	required

Returns:

Name	Type	Description
chunked_list	list[list]	lista di liste, ogni elemento e' una lista con il numero di oggetti indicati provenienti dalla lista madre

Source code in hive/decorators.py

def chunks(lst: list, n: int) -> List[list]:
    """
    metodo per suddividere una lista madre in una nuova lista composta di sotto liste piu piccole

    Args:
        lst (list): lista da suddividere
        n (int): numero di componenti dentro ogni sotto suddivisione

    Returns:
        chunked_list (list[list]): lista di liste, ogni elemento e' una lista con il numero di oggetti indicati
            provenienti dalla lista madre
    """
    chunked_list = []
    for i, ii in enumerate(range(0, len(lst), n)):
        chunked_list.append(lst[ii:ii + n])
    return chunked_list

paginate(single_page, page_size, skip, limit, bulk)

Questa funzione è un decorator che abilita la paginazione per una funzione.

Parameters:

Name	Type	Description	Default
single_page	bool	se True, restituisce solo una singola "pagina" di risultati.	required
page_size	int	dimensione della pagina (numero massimo di elementi per chiamata).	required
skip	int	numero di elementi da saltare all'inizio.	required
limit	int	numero massimo di elementi da restituire.	required
bulk	bool	se True, gestisce il payload in modalità "bulk", suddividendolo in chunk.	required

Source code in hive/decorators.py

def paginate(single_page: bool, page_size: int, skip: int, limit: int, bulk: bool):
    """Questa funzione è un decorator che abilita la paginazione per una funzione.

    Args:
        single_page: se True, restituisce solo una singola "pagina" di risultati.
        page_size: dimensione della pagina (numero massimo di elementi per chiamata).
        skip: numero di elementi da saltare all'inizio.
        limit: numero massimo di elementi da restituire.
        bulk: se True, gestisce il payload in modalità "bulk", suddividendolo in chunk.
    """
    def attributes(func):
        # Questo è il decorator interno che avvolge la funzione passata (`func`).
        @functools.wraps(func)
        def behaviour(mode, url, headers, payload, params, **kwargs) -> list:
            # `behaviour` è la funzione wrapper che implementa la logica di paginazione.
            # Accetta i parametri della funzione originale (`func`) e aggiunge il supporto alla paginazione.
            result = []  # Lista che conterrà i risultati aggregati.

            # se la chiamata è bulk, il payload deve essere un lista e qui viene divisa in chunk e viene chiamata un
            # chunk per volta, ogni chunk ha la dimensione del page_size
            if bulk and not single_page:  # se la bulk viene richiesta in single_page, ricade nell'uso normale
                # Divide il payload in chunk della dimensione specificata da `page_size`.
                c_payload = chunks(payload, page_size)
                # Per ogni chunk:
                for c in c_payload:
                    # Chiama la funzione originale (`func`) con il chunk corrente.
                    result_partial = func(mode, url, headers, c, params, **kwargs)
                    # Se il risultato non è una lista, lo converte in una lista.
                    if not isinstance(result_partial, list): result_partial = [result_partial]
                    # Aggiunge i risultati parziali alla lista complessiva `result`.
                    result += result_partial

            else:  # Se non è in modalità bulk o se è richiesto single_page:
                size = page_size if not single_page else limit  # Determina la dimensione della pagina.
                params['skip'] = skip  # Imposta il valore iniziale di `skip` nei parametri.
                params['limit'] = min(size, limit)  # Imposta il limite massimo per la pagina corrente.
                while True:  # Ciclo per iterare attraverso le pagine.
                    # Chiama la funzione originale (`func`) con i parametri della pagina corrente.
                    result_partial = func(mode, url, headers, payload, params, **kwargs)
                    # Se il risultato non è una lista, lo converte in una lista.
                    if not isinstance(result_partial, list): result_partial = [result_partial]
                    # Aggiunge i risultati parziali alla lista complessiva `result`.
                    result += result_partial
                    # Aggiorna il valore di `skip` per passare alla pagina successiva.
                    params['skip'] = params['skip'] + size
                    # Interrompe il ciclo se:
                    # - Non ci sono più risultati (`result_partial` è vuoto).
                    # - Il numero di risultati è inferiore alla dimensione della pagina.
                    # - Il numero totale di risultati supera il limite specificato.
                    # - È richiesto il single_page.
                    if not result_partial or len(result_partial) < size or len(result) > limit or single_page: break

            # se count è True il result è una lista con dentro una tupla, in questa maniera viene trasmessa solo la tupla
            if params.get('count', False):
                result = result[0]

            return result

        return behaviour

    return attributes

warmstart(func=None, active=True, args_ex=None, kwargs_ex=None, verbose=False)

decoratore che permette di scaricare in locale il risutato di una funzione o metodo e se ne gli argomenti ne il corpo della funzione cambiano vengono restituiti i risultati locali invece di calcolare la funzione. Se ci sono argomenti della funzione chiamata che cambiano ad ogni run ma non modificano il risultato, questi argomenti possono essere esclusi con i parametri args_ex e kwargs_ex.

Parameters:

Name	Type	Description	Default
func		waste key, it is not to be filled	None
active	bool	if False the decorator does nothing, Default to True	True
args_ex	list	list of indexes of the function *args to be excluded for the evaluation of the rerun. Default to None.	None
kwargs_ex	list	list of keys of the function **kwargs to be excluded for the evaluation of the rerun. Default to None.	None
verbose	bool	if True print the status of the warm_start. Default to False	False

Returns: func result

Source code in hive/decorators.py

def warmstart(func=None, active: bool = True, args_ex: list = None, kwargs_ex: list = None, verbose=False):
    """
    decoratore che permette di scaricare in locale il risutato di una funzione o metodo e se ne gli argomenti
    ne il corpo della funzione cambiano vengono restituiti i risultati locali invece di calcolare la funzione.
    Se ci sono argomenti della funzione chiamata che cambiano ad ogni run ma non modificano il risultato, questi
    argomenti possono essere esclusi con i parametri args_ex e kwargs_ex.

    Args:
        func: waste key, it is not to be filled
        active (bool, optional): if False the decorator does nothing, Default to True
        args_ex (list, optional): list of indexes of the function *args to be excluded for the evaluation of the rerun.
            Default to None.
        kwargs_ex (list, optional): list of keys of the function **kwargs to be excluded for the evaluation of the
            rerun. Default to None.
        verbose (bool, optional): if True print the status of the warm_start. Default to False

    Returns: func result

    """
    if not func:
        return lambda f: warmstart(func=f, active=active, args_ex=args_ex, kwargs_ex=kwargs_ex, verbose=verbose)

    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        if active:
            if verbose: print('warm start active')

            directory = 'warmstart'
            in_folder = True
            # si controlla se esiste la cartella warmstart
            if not os.path.isdir(directory):
                # nel caso non si riuscisse a creare una cartella viene usata la posizione dello script per creare i file di warmstart
                in_folder = False
                try:
                    # se non esiste provo a crearne una
                    os.mkdir(directory)
                    # se sono riuscito a creare una cartella da qui in poi viene usata la root della cartella
                    in_folder = True
                except:
                    pass

            # recupero il nome dei file nella cartella
            files_available = os.listdir(directory) if in_folder else os.listdir()
            # hash del corpo e del nome della funzione
            body_and_name = hashlib.md5(inspect.getsource(func).encode()).hexdigest()
            # vengono estratte dalle chiavi per l'hash gli args e kwargs che non devono essere considerati valido
            # per quelle situazioni dove un elemento cambia tra una run e la successiva ma non modifica il risultato
            args_key = list(args).copy()
            kwargs_key = kwargs.copy()

            if args_ex is not None:
                args_val = [args_key[el] for el in args_ex]
                for el in args_val:
                    args_key.remove(el)

            if kwargs_ex is not None:
                for el in kwargs_ex:
                    kwargs_key.pop(el)

            # hash delle chiavi della funzione
            arguments = hashlib.md5((str(kwargs_key) + str(args_key)).encode()).hexdigest()
            # creao un nome che contiene i due hash
            filename = 'temp_' + body_and_name + "_" + arguments + ".pkl"
            # verifico se quel nome e' presente nella cartella
            if filename in files_available:
                if in_folder: filename = './' + directory + '/' + filename
                if verbose: print('read')
                return read(filename)
            else:
                if in_folder: filename = './' + directory + '/' + filename
                if verbose: print('write')
                res = func(*args, **kwargs)
                write(res, filename)
                return res
        else:
            if verbose: print('warm start disabled')
            return func(*args, **kwargs)

    return wrapper