Car quand vous allez publier du code, un script, une lib voire, Dieu ait pitié de vous, un framework, les gens qui vont les utiliser vont avoir des besoins auxquels vous n’aviez pas pensé. Or, ils ne voudront pas modifier votre code. L’interêt d’utiliser le code d’un autre, c’est justement de s’éviter la maintenance. Ils voudront aussi prendre en main rapidement votre outil pour faire des choses simples sans avoir à tout comprendre.

Ainsi, il va vous falloir intégrer divers moyens d’étendre les fonctionnalités de votre code, afin que les autres dev puissent l’utiliser dans le cadre de leurs besoins.

Nous allons donc construire un projet bidon : une barre de progression en ASCII dont la sortie va ressembler à ceci :

Starting [========================================= ] 93%

L’idée est de pouvoir l’utiliser ainsi :

    with ProgressBar() as pb:
        for progres in faire_un_truc():
            pb.progress = progres

Nous allons le proposer ce projet sous forme de classe importable, et nous allons voir plusieurs manières de rendre ce code facile à prendre en main, configurable et extensible :

• Paramètres, avec un ordre intelligent et des valeurs par défaut saines.
• Templating de l’ouput.
• Methodes pouvant être écrasée par héritage.
• Getter and setter (ici via des propriétés)
• Exposition intelligente des attributs de la classe.
• Callbacks et passage réfléchi de paramètres à ceux-ci.

J’aurais pu ajouter l’injection de dépendance, mais j’ai déjà traité ce sujet dans un autre article, et je ne voulais pas charger celui-ci. Il est déjà bien gras.

J’ai aussi volontairement omis la docstring et les comments. Écrire une bonne doc est un article à part entière.

Pré-requis : être parfaitement à l’aise avec le POO et les références, comprendre le principe des context managers, avoir pigé les callbacks.

Et roulez jeunesse !

# -*- coding: utf-8 -*-



import sys

from io import IOBase


# On hérite d'IOBase pour permettre le détournement de stdout. Voir plus bas.
class ProgressBar(IOBase):


    # On met les paramètres par ordre de fréquence d'utilisation. Il est
    # plus probable que les programmeurs changent le total que la sortie
    # du programme.
    # Mettre des valeurs par défaut saines permet à l'utilisateur de faire
    # des essais sans trop regarder la doc. Par ailleurs, des valeurs par
    # défaut servent de documentation en soit, et apparaitront si help()
    # est appelé.
    def __init__(self, total=100,  percent_per_sign=1,
                 progress_sign='=', callbacks=(),
                 template='Starting [{bar}] {progress}%',
                 output=sys.stdout, intercept_stdout=True):

        # Par défaut le total est 100, afin que l'utilisateur passe un
        # pourcentage, ce qui est le plus naturel. Néanmoins, si il souhaite
        # s'affranchir de calculs, on lui permet de passer une autre valeur
        # qui sera ramenée à un pourcentage automatiquement à l'affichage
        # de toute façon.
        self.total = total

        # Customisation de base : changer le symbole de progrès. Par défaut
        # un égal, mais certains aiment les ., les - ou autre.
        self.progress_sign = progress_sign

        # Idem, si la personne veut une barre plus ou moins longue.
        self.percent_per_sign = percent_per_sign

        # la longueur de la barre de progression
        self.bar_len = 100 / percent_per_sign * len(progress_sign)

        # Le formatage de la progress bar se fait à l'aide d'une chaîne
        # ordinaire qui sert de template (avec le langage de formatage de Python)
        # et peut facilement être changée pour obtenir plus de contrôle
        # sur l'aspect de la bar.
        self.template = template

        # On peut aussi passer un tuple de fonctions qui permettent de réagir
        # à chaque fois que le progrès change. On utilise nous-même notre
        # système de callback pour que la methode print_progress() soit
        # appelée à chaque fois, mettant à jour l'affichage de la bar.
        self.callbacks = callbacks + (self.print_progress,)

        # Initialisation de 3 variables à vide. 'buffer' va contenir
        # tout ce qui va être écrit avec write() pour éviter de perturber
        # l'affichage de la bar, et '_progress' va contenir le progrès, mais
        # le '_' signale que c'est une variable à usage interne. Nous allons
        # en effet l'enrober dans une propriété. Enfin cursor_shift est
        # le nombre de caractères à effacer pour redessiner la bar à chaque
        # mise à jour.
        self.buffer = ''
        self._progress = 0
        self.cursor_shift = 0

        # Par défaut, on s'attend à ce que l'utilisateur affiche cette
        # bar directement dans le terminal, sur la sortie standard. Mais
        # il peut vouloir déporter l'affichage ailleurs (par exemple stderr).
        # Pour cette raison, on permet de passer le stream sur lequel
        # l'utilisateur va écrire, même si par défaut on prend sys.stdout,
        # donc la sortie standard.
        # Si l'utilisateur ne souhaite rien de tout cela, il peut également
        # retirer 'self.print_progress' de la liste des callbacks puisque
        # l'attribut est public.
        self.out = output

        # Comme on utilise la sortie standard par défaut, la barre peut
        # facilement être cassée par un autre code écrivant aussi sur la sortie
        # standard. Puisqu'un débutant ne comprendra pas rour se suite ce qui se
        # passe, par défaut on va intercepter stdout et mettre tout ce qui est
        # écrit dessus dans un buffer. Si jamais il y a des prints fait durant
        # l'affichage, il seront cachés, et stockés. Ce comportement n'est pas
        # forcément désirable, et peut donc être désactivé par un paramètre pour
        # l'utilisateur qui sait ce qu'il fait notamment dans le cadre
        # d'utilisation des threads.
        self._intercept_stdout = intercept_stdout and self.out == sys.stdout
        if self._intercept_stdout:
            # L'interception de la sortie standard se fait en remplaçant
            # sys.stdout par soi-même. C'est pour cette raison que ProgressBar
            # hérite de IOBase. En effet, de cette manière, on possède
            # l'interface d'un objet stream et tout écriture sur 'self' semblera
            # fonctionner comme sur sys.stdout. Cela permet de faire des prints
            # ou de lancer un shell et d'utiliser la barre, bien que
            # dans un shell cela monopolisera le prompt.
            sys.stdout = self


    # Afficher la barre à la création de l'objet retire de la flexibilité à
    # l'utilisateur qui peut vouloir le créer d'un côté, le stocker, et
    # démarrer l'affichage plus tard. L'affichage est donc conditionné
    # par cette méthode.
    def show(self):
        bar = self.format()
        self.out.write(bar)
        self.cursor_shift = len(bar)
        return self

    # Comme le plus souvent on voudra tout de même afficher la barre juste après
    # la création, on transforme cette classe en context manager. Pour cela on
    # créer un alias de la method show() qu'on appelle __enter__, puisque tout
    # objet qui a une méthode nommée __enter__ peut être utilisé comme context
    # manager. Si vous ne vous souvenez pas de ce que c'est, il y a un article
    # sur le blog à ce sujet. Mais en résumé, ce sont les objets utilisables
    # avec "with".
    # Notez qu'on ne fait pas __enter__ = show, ce qui serait un moyen plus
    # court d'aliaser, car il empêcherait __enter__ d'appeler le bon show()
    # en cas d'héritage, si show() est écrasé.
    def __enter__(self):
        return self.show()


    # Une petit méthode de nettoyage, qui ici va simplement remettre sys.stdout
    # à sa place.
    def stop(self, *args, **kwargs):
        if self._intercept_stdout:
            sys.stdout = self.out

    # Même topo, on alias stop en __exit__ pour avoir la sortie du context
    # manager. On a pas appelé directement ces méthodes __enter__ et __exit__
    # car l'utilisateur peut vouloir les appeler manuellement.
    def __exit__(self, *args, **kwargs):
        return self.stop()


    # Une simple propriété qui donne accès au progres en lecture...
    @property
    def progress(self):
        return self._progress


    # ... et en écriture. La différence étant qu'à l'écriture, on vérifie
    # que la valeur est bien comprise entre 0 et le total. On appelle aussi
    # les callbacks, et donc l'affichage se mettra à jour.
    @progress.setter
    def progress(self, value):

        if not 0 <= value <= self.total:
            raise ValueError("'value' is %s and should be set between 0 "
                             "and 'total' (%s)" % (value, self.total))

        # On le fait avant car les callbacks doivent être appelés
        # avec un état propre.
        previous_progress = self._progress
        self._progress = value

        # Ce que l'on va passer en paramètres aux callbacks est un choix
        # important. Déjà en premier paramètre, on passe self. Ainsi le callback
        # aura accès à presque tout, et on est certain qu'il ne se trouvera
        # pas dépourvu d'informations. Ensuite on passe le progrès. Normalement
        # il peut le récupérer à travers self.progress, mais comme on sait
        # que c'est une information très utilisée, on la passe par politesse,
        # pour faciliter la vie de l'utilisateur. Enfin, une information qu'il
        # ne pourrait pas avoir autrement est le progrès précédent. Bien que
        # nous ne l'utilisons pas dans notre callback, on peut imaginer que
        # c'est le genre d'info qui peut être utile, et qui ne peut être trouvée
        # dans self.
        for callback in self.callbacks:
            callback(self, self._progress, previous_progress)

        # Si on arrive au bout, on appelle stop() automatiquement. Stop()
        # étant idempotente, ce n'est pas grave si elle est appelée plusieurs
        # fois.
        if value == self.total:
            self.stop()

    # Un cas intéressant : pourquoi on ne fait pas self.template directement au
    # lieu de créer une property à vide ? Tout simplement parce qu'un template
    # est typiquement quelque chose que quelqu'un peut vouloir créer
    # dynamiquement (par exemple pour y ajouter une notion de temps qui passe).
    # Donc, on propose de passer le template en paramètre  dans __init__ car la
    # plupart du temps, c'est juste ce qu'on voudra faire : un simple changement
    # cosmétique statique. Mais afin de permettre plus d'extensibilité, on en
    # fait une propriété, qui, comme toute méthode, peut être écrasée avec de
    # l'héritage et permettrait à l'utilisateur de vraiment personnaliser son
    # affichage de manière poussée.
    @property
    def template(self):
        return self._template

    @template.setter
    def template(self, value):
        self._template = value

    # C'est cette méthode qui est appelée si on a détourné sys.stdout quand
    # l'utilisateur va faire un print. Elle doit s'appeler write(), car c'est
    # l'interface connue des objets stream. Grosso modo, on va juste
    # tout stocker dans une variable plutôt que d'afficher quoique ce soit
    def write(self, value):
        self.buffer += value

    # Pour une progrès donné, retourne la barre de progrès sous forme de
    # string ASCII. Elle est mise à part car elle peut ainsi être facilement
    # utilisée dans un callback.
    def format(self, progress=0):
        progress = progress * 100 / self.total
        bar = progress / self.percent_per_sign * self.progress_sign
        bar += (self.bar_len - len(bar)) * ' '
        return self.template.format(bar=bar, progress=progress)


    # Notre callback que l'on utilise pour afficher la barre. C'est une méthode
    # statique car cela nous permet de nous mettre dans les conditions exacte
    # d'un callback d'un utilisateur, qui ne sera qu'une fonction, sans self.
    @staticmethod
    def print_progress(progress_bar, progress, previous_progress):
        # '\b' permet de reculer le curseur dans le terminal, ce qui va
        # nous permettre de réécrire par dessus l'ancienne bar, et la mettre
        # à jour. Si vous tenez à faire un display multi ligne, sachez que '\b'
        # ne permet pas de reculer sur un saut de ligne, pour ça il faut
        # utiliser '\033[1A'
        progress_bar.out.write(progress_bar.cursor_shift * '\b')
        bar = progress_bar.format(progress)
        progress_bar.out.write(bar)
        # flush est nécessaire pour vider le buffer et obtenir un affichage
        # immédiat quand on écrit en direct sur un stream.
        progress_bar.out.flush()
        # on met à jour l'avancement du curseur, qui nous permettra de reculer
        # d'autant au prochain print_progress()
        progress_bar.cursor_shift = len(bar)



if __name__ == "__main__":

    # voici une utilisation standard de la barre :

    import time

    total = 1000
    # L'utilisation du context manager permet de ne pas se soucier d'appeler
    # show() ou stop() et autre détails d'implémentation.
    # On note aussi qu'avoir des valeurs par défaut pour l'initialisation de la
    # barre la rend très facile à utiliser sans trop se prendre la tête, et
    # ce malgré un code derrière assez complexe. On aurait même pu se passer
    # de "total".
    with ProgressBar(total) as pb:

        for i in range(total):
            time.sleep(0.001)
            # on voir l'interêt d'utiliser une property ici, ça rend la mise
            # à jour du progrès très simple
            pb.progress = i
            if i == total / 2:
                print("\nHalf the work already !")
                half = True
            if i == total * 0.7:
                print("Almost done")
                almost = True

        pb.progress = total

    # On peut afficher tout ce qui a été printé durant la progression de la
    # barre si besoin.
    print(pb.buffer)

    from datetime import datetime

    # Et une utilisation custo où on compte les secondes depuis le départ
    # et on les affiches dans le template

    class MaProgressBar(ProgressBar):

        def show(self):
            self.start = datetime.now()
            return super(MaProgressBar, self).show()

        # Comme on a template en tant que méthode, on peut l'overrider et
        # obtenir un comportement
        @property
        def template(self):
            seconds = (datetime.now() - self.start).seconds
            return '{bar} (running for %s seconds)' % seconds

        @template.setter
        def template(self, value):
            pass

    # Notre système de callback va se trouver utile :
    # on met un callback de plus qui va écrire dans un fichier (mais
    # il pourrait faire n'importe quoi, envoyer un post sur un site Web,
    # un mail, formatter le disque...)
    def update_progress(progress_bar, progress, previous_progress):
        with open('/tmp/progress.log', 'w') as log:
            log.write(str(progress))

    # On change aussi le signe de progrès pour quelque chose de plus pro
    pb = MaProgressBar(progress_sign=':-) ', percent_per_sign=10,
                       callbacks=(update_progress,))

    # Au besoin, on peut passer à l'affichage en manuel.
    pb.show()

    for i in range(100):
        time.sleep(.1)
        pb.progress = i

    pb.progress = 100

    pb.stop()

Télécharger le code de l'article

Ce que je vais vous montrer est mal. Mais c’est tellement bon.

from contextlib import contextmanager

# on fait une exception qui hérite de StopIteration car c'est ce qui est utilisé
# de toute façon pour arrêter une boucle
class MultiStopIteration(StopIteration):
    # la classe est capable de se lever elle même comme exception
    def throw(self):
        raise self


@contextmanager
def multibreak():
    # le seul boulot de notre context manager c'est de donne le moyen de lever
    # l'exception tout en l'attrapant
    try:
        yield MultiStopIteration().throw
    except MultiStopIteration:
        pass

En gros on se créé un petit context manager, dont le seul but est de créer une exception qui va remonter en pêtant toutes les boucles. Je vous avais dit que c’était mal

Ca s’utilise comme ça:

>>> with multibreak() as stop:
...     for x in range(1, 4):
...         for z in range(1, 4):
...             for w in range(1, 4):
...                 print w
...                 if x * z * w == 2 * 2 * 2:
...                     print 'stop'
...                     stop() # appel MultiStopIteration().throw()
...
1
2
3
1
2
3
1
2
3
1
2
3
1
2
stop

Je vous avais dit que ça serait bon.

A part le fait que ce n’est pas très rapide au moment du bubbling de l’exception sur 3 blocks, il n’y a aucun danger ou side-effect. On triche en fait à peine, car le mécanisme interne des boucles en Python utilise de toute façon déjà une exception (StopIteration) pour dire à une boucle quand s’arrêter.

Bref, encore une victoire de connard.

Une notation littérale pour les sets:

>>> {1, 2} == set((1, 2))
True

Une syntaxe pour les dictionnaires en intention:

>>> d = {chr(x): x for x in range(65, 91)}
>>> d
{'A': 65, 'C': 67, 'B': 66, 'E': 69, 'D': 68, 'G': 71, 'F': 70, 'I': 73, 'H': 72, 'K': 75, 'J': 74, 'M': 77, 'L': 76, 'O': 79, 'N': 78, 'Q': 81, 'P': 80, 'S': 83, 'R': 82, 'U': 85, 'T': 84, 'W': 87, 'V': 86, 'Y': 89, 'X': 88, 'Z': 90}

Imbriquer with:

Avant il fallait utiliser nested() ou imbriquer à la main

with open('fichiera') as a:
    with open('fichiera') as b:
        # faire un truc

Maintenant on peut faire:

with open('fichiera') as a, open('fichiera') as b:
    # faire un truc

Rien à voir, mais toujours sympa. timedelta a maintenant une méthode total_seconds() qui retourne la valeur de la durée en seconde. En effet, l’attribut seconds ne retourne que ce qui reste en seconde une fois qu’on a retiré les jours:

>>> from datetime import timedelta
>>> delta = timedelta(days=1, seconds=1)
>>> delta.seconds
1
>>> delta.total_seconds()
86401.0

Notez qu’il n’y a toujours ni attribut minutes, ni heures.

Le module unittest gagne une pléthore d’améliorations, et notamment:

L’utilisation de assertRaises comme context manager:

with self.assertRaises(KeyError):
    {}['foo']

Et un bon gros nombres de méthodes:

assertIsNone() / assertIsNotNone(), assertIs() / assertIsNot(), assertIsInstance() / assertNotIsInstance(), assertGreater() / assertGreaterEqual() / assertLess() / assertLessEqual(), assertRegexpMatches() / assertNotRegexpMatches(), assertRaisesRegexp(),
assertIn() / assertNotIn(), assertDictContainsSubset(), assertAlmostEqual() / assertNotAlmostEqual().

Enfin format() commence à devenir une alternative valable à % car il propose maintenant des marqueurs sans noter d’index:

>>> "{}, puis {} et finalement {}".format(*range(3))
'0, puis 1 et finalement 2'

Et il ajoute le séparateur des milliers au mini-langage de formatage, mais pour la virgule uniquement. Par exemple, si avoir un nombre de 15 caractères minimum formater en tant que float, avec deux chiffres après la virgules, et donc les milliers sont groupés à l’américaine:

>>> '{:15,.2f}'.format(54321)
'      54,321.00'

Heureusement, avec un petit hack, on peut capturer ce qu’affiche un autre code, et sauver le bébé, l’eau du bain, et même le canard en plastique.

Le code pour les gens pressés

J’ai enrobé l’astuce dans un context manager, ça rend l’utilisation plus simple.

import sys
from io import BytesIO
from contextlib import contextmanager

@contextmanager
def capture_ouput(stdout_to=None, stderr_to=None):
    try:

        stdout, stderr = sys.stdout, sys.stderr
        sys.stdout = c1 = stdout_to or BytesIO()
        sys.stderr = c2 = stderr_to or BytesIO()
        yield c1, c2

    finally:

        sys.stdout = stdout
        sys.stderr = stderr

        try:
            c1.flush()
            c1.seek(0)
        except (ValueError, IOError):
            pass

        try:
            c2.flush()
            c2.seek(0)
        except (ValueError, IOError):
            pass

Notez l’usage de yield.

Et ça s’utilise comme ça:

with capture_output() as stdout, stderr:
    fonction_qui_fait_que_printer_la_biatch()

print stdout.read() # on récupère le contenu des prints

Attention, le code n’est pas thread safe, c’est fait pour hacker un code crade, pas pour devenir une institution. Mais c’est fort pratique dans notre cas précis.

Comment ça marche ?

stdin (entrée standard), stdout (sortie standard) et stderr (sortie des erreurs) sont des file like objects, c’est à dire qu’ils implémentent l’interface d’un objet fichier: on peut les ouvrir, les lire, y écrire et les fermer avec des méthodes portant le même nom et acceptant les mêmes paramètres.

L’avantage d’avoir une interface commune, c’est qu’on peut du coup échanger un file like objet par un autre.

Par exemple on peut faire ceci:

import sys
log = open('/tmp/log', 'w')
sys.stdout = log # hop, on hijack la sortie standard
print "Hello"
log.close()

Comme print écrit dans stdout, en remplaçant stdout par un fichier, print va du coup écrire dans le fichier.

Mais ce code est fort dangereux, car il remplace stdout de manière définitive. Du coup, si du code print après, il va écrire dans le fichier, même les libs externes, car stdout est le même pour tout le monde dans le process Python courant.

Du coup, il est de bon ton de s’assurer la restauration de stdout à son état d’origine:

import sys
log = open('/tmp/log', 'w')
bak = sys.stdout # on sauvegarde l'ancien stdout
sys.stdout = log
print "Hello"
log.close()
sys.stdout = bak # on restore stdout

Comme je le disais plus haut, ceci n’est évidement pas thread safe, puisqu’entre la hijacking et la restoration de stdout, un autre thread peut faire un print.

Dans notre context manager, on utilise BytesIO() et non un fichier. BytesIO est un file like objet qui permet de récupérer un flux de bits en mémoire. Donc on fait écrire print dedans, ainsi on a tout ce qu’on affiche qui se sauvegarde en mémoire.

Bien entendu, vous pouvez créé vos propres file like objects, par exemple un objet qui affiche à l’écran ET capture la sortie. Par exemple, pour mitiger le problème de l’absence de thread safe: 99% des libs n’ont pas besoin du vrai stdout, juste d’un truc qui print.

import sys
from io import BytesIO

class PersistentStdout(object):

    old_stdout = sys.stdout

    def __init__(self):
        self.memory = BytesIO()

    def write(self, s):
        self.memory.write(s)
        self.old_stdout.write(s)


old_stdout = sys.stdout
sys.stdout = PersistentStdout()

print "test" # ceci est capturé et affiché

sys.stdout.memory.seek(0)
res = sys.stdout.memory.read()

sys.stdout = PersistentStdout.old_stdout

print res # résultat de la capture

Pour cette raison le code du context manager permet de passer le file like objet à utiliser en argument. On notera aussi que si on souhaite rediriger stdout mais pas stderr et vice-versa, il suffit de passer sys.stdout et sys.stderr en argument :-)