Résultats de notification

notify() et async_notify() retournent un objet résultat plutôt qu’un simple True/False. Pour vérifier seulement la réussite ou l’échec, traitez-le comme un booléen comme avant.

import apprise

# 1. Construisez votre objet Apprise comme d'habitude
apobj = apprise.Apprise()
apobj.add("mailto://user:pass@example.com")

# 2. notify() retourne un objet résultat -- mais vous pouvez toujours
#    le traiter comme un simple True/False si c'est tout ce qu'il vous faut
result = apobj.notify(title="Sauvegarde", body="Terminée avec succès")

if result:
    print("Notification envoyée")

Le reste de cette page montre comment savoir quel service a échoué, pourquoi il a échoué, et combien de temps cela a pris.

Migration depuis une version antérieure

notify() retournait auparavant simplement True, False ou None. Le code qui traite le résultat comme un booléen (if result:) continue de fonctionner sans changement. Le code qui compare par identité, non — result is True, result is False et result is None valent désormais tous False, puisque result est toujours un objet AppriseResult. Remplacez ces vérifications par result.status :

from apprise import AppriseResultStatus

result = apobj.notify(body="Bonjour")

if result:                                            # Fonctionne toujours
    ...

if result.status == AppriseResultStatus.NOMATCH:      # Remplace `result is None`
    ...

La Version Courte

result = apobj.notify(body="Sauvegarde nocturne terminée")

print(bool(result))               # True/False -- même signification qu'avant
print(result.status.name)         # SUCCESS, FAILURE, NOMATCH, PARTIAL ou TIMEOUT
print(len(result))                # combien de services ont réellement été contactés
print(result.success_count)       # combien d'entre eux ont réussi
print(result.failed_count)        # combien n'ont pas réussi

result.status est un AppriseResultStatus (un IntEnum) — l’afficher directement affiche sa valeur entière (par ex. 0), pas son nom. Utilisez .name, ou comparez-le directement à des valeurs comme AppriseResultStatus.SUCCESS, pour obtenir la forme lisible.

Le Résultat Global (status)

result.status peut prendre cinq valeurs. Les nombres correspondent aux codes de sortie de la CLI Apprise, donc les scripts qui vérifient $? peuvent garder les mêmes significations :

Statut	Code de sortie	Signification
SUCCESS	0	Chaque service correspondant a été notifié avec succès.
FAILURE	1	Chaque service correspondant a échoué (aucun n’a réellement délivré la notification), ou la notification n’a pas pu démarrer.
NOMATCH	3	Aucun service ne correspondait au filtre de tag ou de priorité, donc rien n’a été tenté.
PARTIAL	4	Certains services ont envoyé la notification, d’autres non — voir Résultats mixtes par service ci-dessous.
TIMEOUT	5	Rien n’a été envoyé, mais le seul problème était le manque de temps — voir Délais par Service.

from apprise import AppriseResultStatus

# Un filtre de priorité exclusif qui ne correspond à rien retourne un
# résultat qui vaut False avec un statut NOMATCH. Cela signifie
# qu'aucun service ne correspondait.
result = apobj.notify(body="Déploiement terminé", tag="2:alerts")

if result.status == AppriseResultStatus.NOMATCH:
    print("Aucun service ne correspond à cette combinaison tag/priorité")

Résultats mixtes par service

result.status est une seule valeur pour tout l’appel, mais un même appel peut notifier plusieurs services avec des résultats différents. Apprise résume ces résultats en un seul statut de la façon suivante :

1. Tout le monde a réussi ? SUCCESS.
2. Sinon, au moins un service a-t-il vraiment envoyé la notification ? PARTIAL — certains sont passés, d’autres non.
3. Sinon (rien n’a été envoyé), au moins un service a-t-il échoué ? FAILURE — un échec clair est plus utile que de résumer tout l’appel comme un simple dépassement de délai.
4. Sinon, rien n’a été envoyé, mais le seul problème était le temps : TIMEOUT.

result = apobj.notify(body="Diffusion vers trois régions", tag="all")

if result.status == AppriseResultStatus.PARTIAL:
    print("Certaines régions ont été notifiées, d'autres non -- voir le détail par service")
    for service in result:
        print(" ", service.name, bool(service))

Note

Un service optional=yes qui n’a en réalité jamais envoyé la notification ne compte pas comme une vraie réussite pour l’étape 2 ci-dessus, même si son propre service.status (voir Détail par Service ci-dessous) affiche SUCCESS. C’est le comportement attendu de optional. Si un service requis échoue et qu’un service optionnel n’est pas compté comme un échec global, l’appel entier reste FAILURE, pas PARTIAL.

La même règle « un échec confirmé l’emporte sur un délai dépassé » s’applique au sein des tentatives d’un même service : si un service échoue lors d’une tentative puis manque de temps avant qu’une nouvelle tentative puisse démarrer, le status de ce service (ci-dessous) est FAILURE, pas TIMEOUT — l’échec confirmé est plus informatif que « a manqué de temps », donc il l’emporte. TIMEOUT n’apparaît que pour un service qui n’a jamais eu d’échec confirmé du tout (par ex. sa toute première tentative était encore en cours quand le délai a expiré). Si retry vaut 0 (par défaut), il n’y a jamais qu’une seule tentative, donc cette question ne se pose pas — l’ambiguïté ne concerne que les services configurés avec des tentatives.

Détail par Service

Parcourez le résultat pour voir exactement ce que chaque service a fait — « par service » signifie simplement une entrée pour chaque service contacté, comme Slack ou Discord. Chaque entrée est un NotifyResult :

result = apobj.notify(body="Rapport hebdomadaire en pièce jointe")

# Parcourir le résultat renvoie un NotifyResult par service réellement
# contacté (les services ignorés n'apparaissent jamais ici)
for service in result:
    print(service.name, service.url, bool(service), service.status.name)

Champs utiles sur chaque NotifyResult :

Champ	Description
name	Le nom affiché du service, par ex. "Slack".
url	L’URL masquée (confidentialité) qui a été contactée.
url_id	Un identifiant stable pour cette URL qui ne révèle pas ses identifiants.
tag	Tous les tags configurés sur ce service, triés par ordre alphabétique (pas seulement celui du filtre qui l’a fait correspondre cette fois-ci).
status	SUCCESS, FAILURE ou TIMEOUT pour ce seul service — jamais NOMATCH ni PARTIAL, qui ne décrivent tous deux que le lot entier (voir Résultats mixtes par service ci-dessus), pas un service isolé.
optional	Si ce service est marqué optional=yes (voir Services Optionnels).
weight	Le nombre d’appels sous-jacents que ce service est configuré pour effectuer par tentative : son nombre de cibles multiplié par retry + 1 (par ex. une URL SMS avec 50 numéros et sans nouvelle tentative a un poids de 50 ; avec retry=2, c’est 150).
max_attempts	Combien de tentatives étaient autorisées (retry + 1).
elapsed	Secondes entre la première et la dernière tentative de ce service (start_time et end_time sont les horodatages sous-jacents).

Note

Si un service est optional=yes, son status se lit comme SUCCESS même si la livraison réelle a échoué ou dépassé le délai — c’est le comportement prévu de optional. Cela signifie aussi qu’il est compté dans result.success_count (et exclu de result.timeout_count, même si la tentative sous-jacente a effectivement expiré). Regardez les tentatives brutes (ci-dessous) si vous voulez savoir ce qui s’est réellement passé.

Détail par Tentative

Chaque nouvelle tentative (et chaque appel individuel qu’Apprise a effectué pour ce service) est également enregistrée. Parcourez un NotifyResult lui-même pour voir chaque NotifyAttempt :

result = apobj.notify(body="Sauvegarde nocturne terminée")

for service in result:
    # len(service) indique combien de tentatives ce service a réellement utilisées
    print(service.name, "->", len(service), "tentative(s) effectuée(s)")

    # Chaque tentative a son propre statut et sa propre durée. Cela montre
    # quelle tentative a réussi, ou combien de temps un délai dépassé a pris.
    for attempt in service:
        print(" ", attempt.status.name, f"{attempt.elapsed:.2f}s")

Contrairement à service.status, le status d’une tentative est toujours le résultat d’origine — SUCCESS, FAILURE ou TIMEOUT — jamais ajusté pour optional. Chaque tentative a aussi son propre start_time et end_time.

Note

Une tentative TIMEOUT ne représente pas forcément un appel réellement effectué. Quand le délai global expire avant qu’une nouvelle tentative n’ait pu démarrer, Apprise enregistre une tentative TIMEOUT à durée nulle pour marquer cette décision, plutôt que de la faire disparaître silencieusement — donc « chaque tentative est enregistrée » inclut des tentatives qui ont été sautées, pas seulement celles qui ont réellement atteint le service.

Lire les Journaux

Tout avertissement ou erreur qu’un service a journalisé pendant sa notification est capturé et attribué à ce service précis. Cela aide quand plusieurs services sont notifiés en même temps et qu’il faut savoir quel service a produit chaque ligne de journal.

result = apobj.notify(body="Déploiement de la nouvelle version")

for service in result:
    # .logs() renvoie chaque avertissement/erreur journalisé par ce
    # service, dans l'ordre, à travers toutes ses tentatives -- chaque
    # ligne est un NotifyLogEntry
    for line in service.logs():
        print(f"[{service.name}] {line}")

service.logs() est un raccourci qui parcourt toutes les tentatives pour vous. Si vous devez savoir quelle tentative précise a produit quel message, lisez plutôt attempt.logs directement — un itérable de NotifyLogEntry pour cette seule tentative. Chaque NotifyLogEntry a trois champs : level (par ex. "WARNING"), message, et time ; l’afficher avec print() le formate comme une ligne de journal normale (str() reprend le format par défaut du module logging de Python).

NotifyLogEntry prend aussi en charge l’égalité, le hachage et le tri, tous basés sur time (l’égalité/le hachage tiennent aussi compte de level et message). result.logs() — sur l’AppriseResult global, pas service.logs() — s’appuie exactement là-dessus pour vous donner chaque entrée de chaque service déjà fusionnée en une seule chronologie, plutôt qu’un bloc d’entrées par service à la fois :

result = apobj.notify(body="Déploiement de la nouvelle version")

# Chaque entrée de chaque service, rejouée dans l'ordre où elle s'est
# réellement produite
for entry in result.logs():
    print(entry)

Messages non rattachés à un service (call_logs)

Apprise journalise aussi les opérations qui ne concernent pas un service précis, comme les tentatives supplémentaires, l’escalade, une notification qu’il n’a pas pu préparer ou l’absence de service correspondant. Ces entrées sont fournies par result.call_logs() :

result = apobj.notify(body="Déploiement de la nouvelle version")

for entry in result.call_logs():
    print(entry)

Note

Un service qui dépasse son délai est enregistré sur sa propre tentative plutôt qu’ici, si bien que result.logs() ne liste ce dépassement qu’une seule fois. Un log_callback le reçoit malgré tout au moment où il se produit.

result.logs() combine déjà ces entrées avec les journaux des services dans l’ordre chronologique. Utilisez directement call_logs pour consulter uniquement les messages propres à Apprise :

from apprise import AppriseResultStatus

result = apobj.notify(body="test", tag="inexistant")

if result.status == AppriseResultStatus.NOMATCH:
    for entry in result.call_logs():
        print(entry)  # ex. « There are no service(s) to notify »

Capturer plus que les avertissements (log_level)

Sans callback en direct, Apprise capture par défaut le niveau WARNING et les niveaux supérieurs. Avec log_callback, le niveau par défaut devient INFO afin d’inclure les envois réussis. Définissez log_level pour choisir un autre niveau :

import logging

apobj = apprise.Apprise()
apobj.add("mailto://user:pass@example.com")

# Capture aussi les messages de niveau INFO pour cet appel
result = apobj.notify(body="Déploiement de la nouvelle version", log_level=logging.INFO)

for entry in result.logs():
    print(entry)

Définissez log_level sur l’objet Apprise, ou remplacez-le pour un seul appel :

# Chaque appel notify()/async_notify() effectué avec cet objet capture INFO+
apobj = apprise.Apprise(log_level=logging.INFO)

# ...sauf cet appel, qui ne veut que WARNING+ (la valeur par défaut)
apobj.notify(body="Vérification silencieuse", log_level=logging.WARNING)

Par défaut, Apprise conserve toutes les entrées capturées en mémoire. Les applications qui produisent beaucoup de journaux peuvent définir result_log_memory_size et result_log_disk_size dans AppriseAsset afin de transférer les journaux vers un stockage temporaire sur disque. Ces limites couvrent toute la notification, avec tous ses services et ses tentatives.

asset = apprise.AppriseAsset(
    result_log_memory_size=2 * 1024 * 1024,
    result_log_disk_size=256 * 1024 * 1024,
)
apobj = apprise.Apprise(asset=asset)

Les fichiers temporaires sont fermés automatiquement lorsque le résultat est libéré. Appelez result.close() si vous conservez longtemps l’objet, ou utilisez-le comme gestionnaire de contexte :

with apobj.notify(body="Déploiement terminé") as result:
    for entry in result.logs():
        print(entry)

Attention

Le logger Python apprise doit aussi autoriser le niveau demandé. Si les entrées INFO ou DEBUG manquent, vérifiez sa configuration :

import logging

# Autorise les entrées INFO à atteindre la capture d'Apprise
logging.getLogger("apprise").setLevel(logging.INFO)

Les applications qui acceptent un niveau par requête doivent configurer le logger une seule fois, puis utiliser log_level pour chaque appel. Modifier le logger partagé pendant une requête peut perturber d’autres notifications.

Suivre les Journaux en Direct

Tout ce qui précède lit les journaux après la fin de notify(). Pour une console ou un panneau de progression en direct, vous pouvez recevoir chaque entrée pendant l’envoi des notifications.

Utilisez log_callback pour cela. Donnez-lui une fonction, et Apprise l’appelle avec (entry, service) pour chaque NotifyLogEntry au moment où elle est capturée, en direct :

def on_log(entry, service):
    print(f"[{service.service_name if service else 'apprise'}] {entry}")

apobj = apprise.Apprise(log_callback=on_log)
apobj.add("slack://tokenA/tokenB/tokenC")
apobj.add("discord://webhook_id/webhook_token")

# on_log() reçoit chaque entrée capturée avant la fin de notify()
apobj.notify(body="Déploiement de la nouvelle version")

Le callback peut être n’importe quel objet appelable qui accepte ces deux arguments : une fonction, une lambda, ou un objet avec une méthode __call__(). Une fonction classique suffit souvent. Utilisez Apprise(log_callback=...) comme valeur par défaut pour tout l’objet, ou notify(log_callback=...) pour un seul envoi :

recent_logs = []

def collect_log(entry, service):
    # Gardez juste les informations utiles pour la page d'état de votre app.
    recent_logs.append(
        {
            "service": service.service_name if service else "apprise",
            "level": entry.level,
            "message": entry.message,
        }
    )

apobj = apprise.Apprise()
apobj.add("mailto://user:pass@example.com")

# Utilisez ce callback seulement pour cette notification.
result = apobj.notify(
    body="Sauvegarde terminée",
    log_callback=collect_log,
)

log_callback doit être synchrone. S’il renvoie une coroutine, Apprise la ferme sans l’exécuter et journalise un avertissement. Pour publier de façon asynchrone, planifiez le travail sur la boucle d’événements de votre application depuis un callback synchrone :

import asyncio

loop = asyncio.get_event_loop()

def publish_log(entry, service):
    asyncio.run_coroutine_threadsafe(
        websocket.send_json(
            {
                "service": service.service_name if service else "apprise",
                "level": entry.level,
                "message": entry.message,
            }
        ),
        loop,
    )

apobj = apprise.Apprise(log_callback=publish_log)
apobj.add("discord://webhook_id/webhook_token")

result = await apobj.async_notify(body="Déploiement démarré")

Quelques points à connaître :

• C’est entièrement optionnel. Omettez-le (le comportement par défaut) et tout fonctionne exactement comme avant — log_callback ne change rien à la valeur de retour de notify() ni aux journaux que vous pouvez déjà lire dans le résultat après coup.

• Il ne se déclenche que pour ce qui est réellement capturé. log_callback reçoit les entrées autorisées par log_level. Avec un callback, le niveau par défaut est INFO; définissez-le pour utiliser DEBUG, TRACE ou WARNING.

• service vaut None pour une entrée de call_logs. Gérez ce cas, par exemple : service.service_name if service else "apprise".

• Définissez-le une fois, ou juste pour un appel. Apprise(log_callback=...) s’applique à chaque appel notify()/async_notify() effectué avec cet objet. Passer log_callback= directement à notify() remplace cette valeur par défaut pour ce seul appel :

  apobj.notify(body="Alerte ponctuelle", log_callback=on_log)

• Il peut être appelé depuis plusieurs threads à la fois. Apprise peut notifier plusieurs services en parallèle, donc si deux services journalisent un avertissement au même moment, on_log() peut réellement s’exécuter deux fois en même temps, sur deux threads différents. Gardez le callback court et évitez de toucher un état partagé à moins qu’il ne soit thread-safe (une queue.Queue l’est ; une simple liste sur laquelle vous faites append() sans verrou ne l’est pas).

• Il doit être synchrone. Apprise n’exécute pas les callbacks asynchrones. Planifiez le travail asynchrone depuis un callback synchrone, comme ci-dessus.

• Si votre callback lève une exception, notify() continue quand même. L’erreur est journalisée, pas relancée vers vous — un bug dans on_log() ne devrait jamais pouvoir casser une notification réelle.

Compter les Choses

result = apobj.notify(body="Diffusion sur tous les canaux", tag="all")

print(len(result))            # services réellement tentés
print(result.success_count)   # combien ont réussi
print(result.failed_count)    # combien n'ont pas réussi
print(result.timeout_count)   # combien ont spécifiquement expiré

Export en JSON

Le résultat complet, un service ou une tentative peut être écrit en JSON sans recharger tous les journaux capturés en mémoire. Utilisez write_json() pour écrire dans un fichier texte ou un autre objet doté d’une méthode write() :

result = apobj.notify(body="Sauvegarde nocturne terminée")

# Écrire chaque service, tentative et entrée de journal directement dans un fichier.
with open("resultat.json", "w", encoding="utf-8") as stream:
    result.write_json(stream)

# Un service ou une tentative prend en charge la même méthode.
for service in result:
    with open("service.json", "w", encoding="utf-8") as stream:
        service.write_json(stream)

Utilisez iter_json() lorsqu’un framework ou un client réseau accepte des blocs :

for chunk in result.iter_json():
    envoyer_au_client(chunk)

L’itérateur garde les journaux stockés sur disque en lecture progressive et produit un document JSON valide du début à la fin. Gardez le résultat ouvert jusqu’à la fin de l’itération. Il est possible de réunir les blocs avec "".join(result.iter_json()) pour un petit résultat, mais cela charge le document JSON complet en mémoire.

Les résultats conteneurs ne proposent volontairement pas asdict() ni une méthode json() qui retourne une chaîne, car ces deux formes doivent charger tous les journaux imbriqués. Les objets NotifyLogEntry, qui restent petits, prennent toujours en charge ces deux méthodes.

À propos de async_notify()

async_notify() retourne exactement le même type d’AppriseResult que notify() — pensez à bien récupérer (et, si le résultat vous importe, vérifier) sa valeur de retour :

result = await apobj.async_notify(body="Sauvegarde nocturne terminée")

if not result:
    print("Quelque chose a échoué :", result.status.name)

Référence Rapide

Vous voulez savoir…	Vérifiez ceci
Est-ce que tout a fonctionné ?	bool(result)
Pourquoi l’appel entier n’a-t-il pas réussi ?	result.status (comparez-le, ou utilisez .name pour l’afficher)
Quels services ont même été tentés ?	for service in result: ...
Un service précis a-t-il réussi ?	bool(service)
Combien de tentatives un service a-t-il pris ?	len(service)
Que s’est-il réellement passé à chaque tentative ?	for attempt in service: ...
Quels avertissements/erreurs un service a-t-il journalisés ?	service.logs() / attempt.logs
Qu’est-ce qu’Apprise lui-même a journalisé (tentatives, escalade) ?	result.call_logs()
Je veux les journaux de tous les services, fusionnés dans l’ordre chronologique	result.logs() (inclut aussi call_logs)
Je veux capturer plus (ou moins) que les avertissements	log_level= sur Apprise() ou notify()
Je veux voir les journaux en direct, au fur et à mesure	log_callback= sur Apprise() ou notify()
Combien de temps cela a-t-il pris ?	.elapsed sur result, service, ou attempt
J’ai besoin d’un JSON à mémoire limitée	.write_json() / .iter_json()

Voir Aussi

• Routage par Tags & Tentatives — escalade par priorité, tentatives/attente par service, services optionnels, et les réglages de délai qui produisent un statut TIMEOUT.
• Assets & Branding — où service_timeout et les autres valeurs par défaut de session sont configurés.
• Diffusion en direct de la progression — utilisez log_callback via l’API Apprise sur HTTP.