Invoke est un ordonnanceur et un exécuteur de tâches, c'est çà dire que il permet de décrire un workflow d'étapes pour atteindre un objectif et l'outil exécute celles-ci dans le bon ordre.

Il repose sur la définition des tâches (task) via l'annotation @task. Ces tâches sont définies au sein d'un fichier tasks.py.

tasks.py

from invoke import task
 
@task
def remove_all(ctx):
   ctx.run('rm -rf /tmp/my_dirs/work')

$ invoke -l
Available tasks:
 
  remove-all
 
$ invoke remove-all
$

Création de l'environnement virtuel de test

$ ~/Workdirs$ virtualenv -p python3 test_invoke
Running virtualenv with interpreter /usr/bin/python3
Using base prefix '/usr'
New python executable in /home/mdexet/Workdirs/test_invoke/bin/python3
Also creating executable in /home/mdexet/Workdirs/test_invoke/bin/python
Installing setuptools, pip, wheel...done.
$ ~/Workdirs$ cd test_invoke/
$ ~/Workdirs/test_invoke$ source ./bin/activate

Installation de pyInvoke

$ ~/Workdirs/test_invoke$ pip install invoke
Collecting invoke
  Downloading invoke-0.21.0-py3-none-any.whl (153kB)
    100% |████████████████████████████████| 153kB 3.9MB/s 
Installing collected packages: invoke
Successfully installed invoke-0.21.0

Pour créer une tâche il suffit

• de créer un fichier tasks.py
• de créer une fonction qui prend au moins un contexte ctx en argument

  • et de l'annoter avec @task

  <code python> @task def do_something(ctx): ….
  </code>

Une tâche peut exécuter une fonction python ou une commande système avec ctx.run(<commande>).

Appel de fonction

def do_it_right():
   //....
 
@task
def do_something(ctx):
   do_it_right()
 

Appel de commande système

@task(pre=[clean_workdir, get_last_commit, compile_project], 
      post=[deploy_artefact, update_website])
def build(ctx):
   ctx.run('maven install')

Un tâche peut nécessiter plusieurs taches en amont et/ou en aval du processus.

Pour modéliser cet enchaînement, il faut indiquer à la tache quels sont

• ses prédécesseurs avec la propriété pre

  • et ses successeurs avec la propriété post

  <code python> @task(pre=[cleanworkdir, getlastcommit, compileproject], post=[deployartefact, updatewebsite]) def build(ctx): do_something() </code>

La succession des tâches s'effectue de la première à la dernière sauf si

• une exception est levée
• une commande système ( ctx.run(<commande>) ) retourne un code erreur.

Exemple d'exception non gérée

@task
def get_last_commit(ctx):
    a = 10/0
    print('Get last commit')

$ invoke build 
Clean workdir
Get last commit
Traceback (most recent call last):
  File "/home/user/Workdirs/test_invoke/bin/invoke", line 11, in <module>
    sys.exit(program.run())
...
  File "/home/user/Workdirs/test_invoke/tasks.py", line 10, in get_last_commit
    a = 10/0
ZeroDivisionError: division by zero

Exemple d'exception gérée

@task
def get_last_commit(ctx):
    raise Exit('hostname not found')
    print('Get last commit')

$ invoke build 
Clean workdir
hostname not found

Pour désigner une tâche par défaut, il suffit de lui donner la propriété default=True

@task(default=True)
def build(ctx):
 ...

from invoke import task
 
@task
def clean_workdir(ctx):
    print('Clean workdir')
 
@task
def get_last_commit(ctx):
    print('Get last commit')
 
@task
def compile_project(ctx):
    print('Commit')
 
@task
def deploy_artefact(ctx):
    print('Deploy')
 
@task
def update_website(ctx):
    print('Update website')
 
@task(  pre=[clean_workdir, get_last_commit, compile_project],
        post=[deploy_artefact, update_website])
def build(ctx):
   print('Building')

$ invoke build
Clean workdir
Get last commit
Commit
Building
Deploy
Update website

Invoke est un ordonnanceur et un exécuteur de tâches.

Il repose sur la définition des tâches (task) via l'annotation @task. Ces tâches sont définies au sein d'un fichier tasks.py.

tasks.py

from invoke import task
 
@task
def remove_all(ctx):
   ctx.run('rm -rf /tmp/my_dirs/work')

$ invoke -l
Available tasks:
 
  remove-all
 
$ invoke remove-all
$

Création de l'environnement virtuel de test

$ ~/Workdirs$ virtualenv -p python3 test_invoke
Running virtualenv with interpreter /usr/bin/python3
Using base prefix '/usr'
New python executable in /home/mdexet/Workdirs/test_invoke/bin/python3
Also creating executable in /home/mdexet/Workdirs/test_invoke/bin/python
Installing setuptools, pip, wheel...done.
$ ~/Workdirs$ cd test_invoke/
$ ~/Workdirs/test_invoke$ source ./bin/activate

Installation de pyInvoke

$ ~/Workdirs/test_invoke$ pip install invoke
Collecting invoke
  Downloading invoke-0.21.0-py3-none-any.whl (153kB)
    100% |████████████████████████████████| 153kB 3.9MB/s 
Installing collected packages: invoke
Successfully installed invoke-0.21.0

Pour créer une tâche il suffit

• de créer un fichier tasks.py
• de créer une fonction qui prend au moins un contexte ctx en argument

  • et de l'annoter avec @task

  <code python> @task def do_something(ctx): ….
  </code>

Une tâche peut exécuter une fonction python ou une commande système avec ctx.run(<commande>).

Appel de fonction

def do_it_right():
   //....
 
@task
def do_something(ctx):
   do_it_right()
 

Appel de commande système

@task(pre=[clean_workdir, get_last_commit, compile_project], 
      post=[deploy_artefact, update_website])
def build(ctx):
   ctx.run('maven install')

Un tâche peut nécessiter plusieurs taches en amont et/ou en aval du processus.

Pour modéliser cet enchaînement, il faut indiquer à la tache quels sont

• ses prédécesseurs avec la propriété pre

  • et ses successeurs avec la propriété post

  <code python> @task(pre=[cleanworkdir, getlastcommit, compileproject], post=[deployartefact, updatewebsite]) def build(ctx): do_something() </code>

La succession des tâches s'effectue de la première à la dernière sauf si

• une exception est levée
• une commande système ( ctx.run(<commande>) ) retourne un code erreur.

Exemple d'exception non gérée

@task
def get_last_commit(ctx):
    a = 10/0
    print('Get last commit')

$ invoke build 
Clean workdir
Get last commit
Traceback (most recent call last):
  File "/home/user/Workdirs/test_invoke/bin/invoke", line 11, in <module>
    sys.exit(program.run())
...
  File "/home/user/Workdirs/test_invoke/tasks.py", line 10, in get_last_commit
    a = 10/0
ZeroDivisionError: division by zero

Exemple d'exception gérée

@task
def get_last_commit(ctx):
    raise Exit('hostname not found')
    print('Get last commit')

$ invoke build 
Clean workdir
hostname not found

Pour désigner une tâche par défaut, il suffit de lui donner la propriété default=True

@task(default=True)
def build(ctx):
 ...

from invoke import task
 
@task
def clean_workdir(ctx):
    print('Clean workdir')
 
@task
def get_last_commit(ctx):
    print('Get last commit')
 
@task
def compile_project(ctx):
    print('Commit')
 
@task
def deploy_artefact(ctx):
    print('Deploy')
 
@task
def update_website(ctx):
    print('Update website')
 
@task(  pre=[clean_workdir, get_last_commit, compile_project],
        post=[deploy_artefact, update_website])
def build(ctx):
   print('Building')

$ invoke build
Clean workdir
Get last commit
Commit
Building
Deploy
Update website

Imaginons un projet avec

• un répertoire main/doc contenant la documentation en asciidoc
• un répertoire build/doc contenant la documentation finale
• un listing de fichiers dans main/src à mettre sous forme de document asciidoc

Décomposons le workflow.

Il faut

* générer un document //_listing.adoc// avec la liste des fichiers dans //main/src//
* générer un PDF dans //build/doc//
   * avant il faut s'assurer que //build/doc// existe

TBD

Si une tâche doit retourner un résultat pour la tâche suivante, comment le passer ? Apparemment il n'est pas possible de transmettre un résultat par l'intermédiaire du contexte.

Le seul moyen est de peut-être utiliser un dictionnaire de stockage global.

storage = {}
 
@task
def do_something(ctx):
   storage['foo'] = 'bar'
 
 
@task(pre=[do_something])
def use_something(ctx):
    bar = storage['foo']

Exemple tiré de la documentation officielle

Supposons que nous ayons une commande qui demande des informations de façon interactive :

$ excitable-program
When you give the OK, I'm going to do the things. All of them!!
Are you ready? [Y/n] y
OK! I just did all sorts of neat stuff. You're welcome! Bye!

Pour cela il faut utiliser un Responder qui va surveiller la question et y répondre.

responder = Responder(pattern=r"Are you ready? \[Y/n\] ", response="y\n")
ctx.run("excitable-program", watchers=[responder])

Si une tâche doit retourner un résultat pour la tâche suivante, comment le passer ? Apparemment il n'est pas possible de transmettre un résultat par l'intermédiaire du contexte.

Le seul moyen est de peut-être utiliser un dictionnaire de stockage global.

storage = {}
 
@task
def do_something(ctx):
   storage['foo'] = 'bar'
 
 
@task(pre=[do_something])
def use_something(ctx):
    bar = storage['foo']

Exemple tiré de la documentation officielle

Supposons que nous ayons une commande qui demande des informations de façon interactive :

$ excitable-program
When you give the OK, I'm going to do the things. All of them!!
Are you ready? [Y/n] y
OK! I just did all sorts of neat stuff. You're welcome! Bye!

Pour cela il faut utiliser un Responder qui va surveiller la question et y répondre.

responder = Responder(pattern=r"Are you ready? \[Y/n\] ", response="y\n")
ctx.run("excitable-program", watchers=[responder])