Notification Results
notify() and async_notify() return a result object instead of a plain
True/False. For a success/fail check, treat it like a boolean as before.
import apprise
# 1. Build your Apprise object as usualapobj = apprise.Apprise()apobj.add("mailto://user:pass@example.com")
# 2. notify() returns a result object -- but you can still just# treat it as True/False if that's all you needresult = apobj.notify(title="Backup", body="Completed successfully")
if result: print("Notification sent")The rest of this page shows how to inspect which service failed, why it failed, and how long it took.
The Short Version
Section titled “The Short Version”result = apobj.notify(body="Nightly backup finished")
print(bool(result)) # True/False -- same meaning as beforeprint(result.status.name) # SUCCESS, FAILURE, NOMATCH, PARTIAL, or TIMEOUTprint(len(result)) # how many services were actually contactedprint(result.success_count) # how many of those succeededprint(result.failed_count) # how many of those did notresult.status itself is an AppriseResultStatus (an IntEnum) — printing it
directly prints its integer value (e.g. 0), not its name. Use .name, or compare
it directly against values such as AppriseResultStatus.SUCCESS, when you want the
readable form.
The Overall Outcome (status)
Section titled “The Overall Outcome (status)”result.status is one of five values. The numbers match the Apprise CLI exit
codes, so scripts checking $? can use the same meanings:
| Status | Exit code | Meaning |
|---|---|---|
SUCCESS | 0 | Every matched service was notified successfully. |
FAILURE | 1 | Every matched service failed (none of them genuinely delivered), or the notification could not start. |
NOMATCH | 3 | Nothing matched the tag or priority filter, so no service was attempted. |
PARTIAL | 4 | Some services delivered and some did not — see Mixed Service Results below. |
TIMEOUT | 5 | Nothing delivered, but the only problem was that services ran out of time — see Per-Service Timeouts. |
from apprise import AppriseResultStatus
# An exclusive priority filter that matches nothing gives back a# result that evaluates as False with a NOMATCH status. This means# there was no matching service to notify.result = apobj.notify(body="Deploy finished", tag="2:alerts")
if result.status == AppriseResultStatus.NOMATCH: print("No service matched that tag/priority combination")Mixed Service Results
Section titled “Mixed Service Results”result.status is one value for the whole call, but a single call can notify
several services with different outcomes. Apprise reduces those outcomes to one
status like this:
- Everyone succeeded?
SUCCESS. - Otherwise, did at least one service really deliver?
PARTIAL— some got through, some did not. - Otherwise (nothing delivered at all), did at least one service fail?
FAILURE— a clear failure is more useful than calling the whole run a timeout. - Otherwise, nothing delivered, but the only problem was time:
TIMEOUT.
result = apobj.notify(body="Broadcast to three regions", tag="all")
if result.status == AppriseResultStatus.PARTIAL: print("Some regions were notified, some weren't -- check per-service detail") for service in result: print(" ", service.name, bool(service))The same “confirmed failure beats timeout” rule applies within one service’s
own retries, too: if a service fails outright on one attempt and then runs out
of time before a further retry can start, that service’s own status (see
below) is FAILURE, not TIMEOUT — the confirmed failure is more
informative than “ran out of time”, so it wins. TIMEOUT only shows up on a
service that never got a confirmed failure at all (e.g. its very first
attempt was still in flight when the deadline hit). If retry is 0 (the
default), there’s only ever one attempt, so this doesn’t come up — the
ambiguity is specific to services configured with retries.
Per-Service Detail
Section titled “Per-Service Detail”Loop over the result to see exactly which service did what — “per-service” here
just means one entry for each service that was contacted, such as Slack or
Discord. Each entry is a NotifyResult:
result = apobj.notify(body="Weekly report attached")
# Iterating the result walks one NotifyResult per service that was# actually contacted (skipped services never show up here)for service in result: print(service.name, service.url, bool(service), service.status.name)Useful fields on each NotifyResult:
| Field | Description |
|---|---|
name | The service’s display name, e.g. "Slack". |
url | The privacy-masked URL that was contacted. |
url_id | A stable, short identifier for this URL that does not reveal its credentials. |
tag | Every tag configured on this service, alphabetically sorted (not just whichever tag/priority token in your filter caused it to be matched this time). |
status | SUCCESS, FAILURE, or TIMEOUT for just this one service — never NOMATCH or PARTIAL, both of which only ever describe the whole batch (see Mixed Service Results above), not one service on its own. |
optional | Whether this service is marked optional=yes (see Optional Services). |
weight | The number of underlying calls this service is configured to make per attempt: its target count multiplied by retry + 1 (e.g. one SMS URL with 50 phone numbers and no retry has a weight of 50; with retry=2 it’s 150). |
max_attempts | How many tries were allowed (retry + 1). |
elapsed | Seconds between this service’s first and last attempt (start_time and end_time are the underlying timestamps). |
Per-Attempt Detail
Section titled “Per-Attempt Detail”Every retry (and every individual call Apprise made for that service) is recorded
too. Loop over a NotifyResult itself to see each NotifyAttempt:
result = apobj.notify(body="Nightly backup finished")
for service in result: # len(service) is how many attempts this one service actually used print(service.name, "->", len(service), "attempt(s) made")
# Each attempt has its own status and timing. This shows # which retry succeeded, or how long a timeout took. for attempt in service: print(" ", attempt.status.name, f"{attempt.elapsed:.2f}s")Unlike service.status, an attempt’s status is always the original result —
SUCCESS, FAILURE, or TIMEOUT — never adjusted for optional. Each attempt
also has its own start_time and end_time.
Reading the Logs
Section titled “Reading the Logs”Any warning or error a service logged while it was being notified is captured and attributed to that specific service. This helps when several services are notified at once and you need to know which service produced each log line.
result = apobj.notify(body="Deploying new release")
for service in result: # .logs() yields every warning/error that service logged, in order, # across all of its attempts -- each one is a NotifyLogEntry for line in service.logs(): print(f"[{service.name}] {line}")service.logs() is a convenience that walks every attempt for you. If you need
to know which specific retry produced which message, read attempt.logs
directly instead — an iterable of NotifyLogEntry for just that one attempt.
Each NotifyLogEntry has three fields: level (e.g. "WARNING"), message,
and time; print()-ing one formats it like a normal log line (str() mirrors
Python logging’s own default format).
NotifyLogEntry also supports equality, hashing, and sorting, all based on
time (equality/hashing also considers level and message). result.logs()
— on the overall AppriseResult, not service.logs() — uses exactly this to
give you every entry from every service already merged into one chronological
timeline, rather than one service’s block of entries at a time:
result = apobj.notify(body="Deploying new release")
# Every entry from every service, replayed in the order it actually happenedfor entry in result.logs(): print(entry)Messages Not Tied to One Service (call_logs)
Section titled “Messages Not Tied to One Service (call_logs)”Apprise also logs work that does not belong to one service, such as retries,
escalation, a notification it could not prepare, or having no matching services.
These entries are in result.call_logs():
result = apobj.notify(body="Deploying new release")
for entry in result.call_logs(): print(entry)result.logs() already combines these entries with all service logs in time
order. Use call_logs directly when you only need Apprise’s own messages:
from apprise import AppriseResultStatus
result = apobj.notify(body="test", tag="nonexistent")
if result.status == AppriseResultStatus.NOMATCH: for entry in result.call_logs(): print(entry) # e.g. "There are no service(s) to notify"Capturing More Than Warnings (log_level)
Section titled “Capturing More Than Warnings (log_level)”Without a live callback, Apprise captures WARNING and higher by default. With
log_callback, the default is INFO so successful deliveries also appear.
Set log_level explicitly when you need a different amount of detail:
import logging
apobj = apprise.Apprise()apobj.add("mailto://user:pass@example.com")
# Also capture INFO-level chatter for this one callresult = apobj.notify(body="Deploying new release", log_level=logging.INFO)
for entry in result.logs(): print(entry)Set log_level on the Apprise object, or override it for one call:
# Every notify()/async_notify() call made with this object captures INFO+apobj = apprise.Apprise(log_level=logging.INFO)
# ...except this one call, which only wants WARNING+ (the default)apobj.notify(body="Quiet check-in", log_level=logging.WARNING)By default, Apprise keeps all captured entries in memory. Applications that
capture large volumes can set result_log_memory_size and
result_log_disk_size on AppriseAsset to spill logs to temporary disk
storage. The limits cover the complete notification, including every service
and retry.
asset = apprise.AppriseAsset( result_log_memory_size=2 * 1024 * 1024, result_log_disk_size=256 * 1024 * 1024,)apobj = apprise.Apprise(asset=asset)Temporary files close automatically when the result is released. Call
result.close() when keeping a result object for a long time, or use it as a
context manager:
with apobj.notify(body="Deploy complete") as result: for entry in result.logs(): print(entry)Watching Logs Live
Section titled “Watching Logs Live”Everything above reads logs after notify() has already finished. If you’re
building a live console or progress panel, you can receive each captured entry
while services are still being notified.
Use log_callback for this. Give it a function, and Apprise calls it with
(entry, service) for every NotifyLogEntry as it’s captured, live:
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() receives each captured entry before notify() returnsapobj.notify(body="Deploying new release")The callback can be any callable that accepts those two arguments: a function, a
lambda, or an object with a __call__() method. A plain function is often enough.
Use Apprise(log_callback=...) as the default for the whole object, or
notify(log_callback=...) for just one send:
recent_logs = []
def collect_log(entry, service): # Save just enough information for your app's own status page. 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")
# Use this callback only for this one notification.result = apobj.notify( body="Backup finished", log_callback=collect_log,)log_callback must be synchronous. If it returns a coroutine, Apprise closes
it without running it and logs a warning. To publish asynchronously, schedule
work on your application’s event loop from a synchronous callback:
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="Deploy started")A few things worth knowing:
-
It’s entirely optional. Leave it out (the default) and everything behaves exactly as before —
log_callbackchanges nothing aboutnotify()’s return value or the logs you can already read from the result afterward. -
It only fires for what’s actually being captured.
log_callbacksees the entries allowed bylog_level. Its default isINFOwhen a callback is active; set it explicitly for DEBUG, TRACE, or WARNING. -
serviceisNonefor acall_logsentry. Not every message is tied to one service, so handle that case, for example:service.service_name if service else "apprise". -
Set it once, or just for one call.
Apprise(log_callback=...)applies to everynotify()/async_notify()call made with that object. Passinglog_callback=directly tonotify()itself overrides that default for just that one call:apobj.notify(body="One-off alert", log_callback=on_log) -
It may be called from more than one thread at once. Apprise can notify several services in parallel, so if two services log a warning at the same moment,
on_log()can genuinely be running twice at once, on two different threads. Keep the callback short, and avoid touching shared state unless it’s thread-safe (aqueue.Queueis; a plain list you’re appending to without a lock is not). -
It must be synchronous. Apprise does not run async callbacks. Schedule async work from a synchronous callback, as shown above.
-
If your callback raises, notify() keeps going. The error is logged, not raised back at you — a bug in
on_log()should never be able to break a real notification.
Counting Things Up
Section titled “Counting Things Up”result = apobj.notify(body="Broadcast to all channels", tag="all")
print(len(result)) # services actually attemptedprint(result.success_count) # how many succeededprint(result.failed_count) # how many did notprint(result.timeout_count) # how many specifically timed outExporting as JSON
Section titled “Exporting as JSON”The whole result, one service, or one attempt can be written as JSON without
loading every captured log back into memory. Use write_json() when writing to
a text file or another object with a write() method:
result = apobj.notify(body="Nightly backup finished")
# Write every service, attempt, and log entry directly to a file.with open("result.json", "w", encoding="utf-8") as stream: result.write_json(stream)
# A service or attempt supports the same method.for service in result: with open("service.json", "w", encoding="utf-8") as stream: service.write_json(stream)Use iter_json() when a framework or network client accepts chunks:
for chunk in result.iter_json(): send_to_client(chunk)The iterator keeps disk-backed logs lazy and produces valid JSON from beginning
to end. Keep the result open until the iterator finishes. Joining the chunks,
such as "".join(result.iter_json()), is possible for small results but loads
the complete JSON document into memory.
Container results intentionally do not provide asdict() or a string-returning
json() method because either form must load every nested log. Individual
NotifyLogEntry objects remain small and still support both methods.
Notes On async_notify()
Section titled “Notes On async_notify()”async_notify() returns the exact same kind of AppriseResult as notify() —
remember to actually capture (and, if you care about the outcome, check) its
return value:
result = await apobj.async_notify(body="Nightly backup finished")
if not result: print("Something failed:", result.status.name)Quick Reference
Section titled “Quick Reference”| You want to know… | Check this |
|---|---|
| Did everything work? | bool(result) |
| Why did the whole call not succeed? | result.status (compare it, or use .name to print it) |
| Which services were even attempted? | for service in result: ... |
| Did one specific service succeed? | bool(service) |
| How many tries did a service take? | len(service) |
| What actually happened on each try? | for attempt in service: ... |
| What warnings/errors did a service log? | service.logs() / attempt.logs |
| What did Apprise itself log (retries, escalation)? | result.call_logs() |
| I want every service’s logs, merged in time order | result.logs() (includes call_logs too) |
| I want to capture more (or less) than warnings | log_level= on Apprise() or notify() |
| I want to see logs live, as they happen | log_callback= on Apprise() or notify() |
| How long did it take? | .elapsed on result, service, or attempt |
| I need this as bounded JSON | .write_json() / .iter_json() |
See Also
Section titled “See Also”- Tag Routing & Retry — priority escalation, per-service
retry/wait, optional services, and the timeout settings that drive a
TIMEOUTstatus. - Assets & Branding — where
service_timeoutand other session-wide defaults are configured. - Live Progress Streaming — use
log_callbackthrough Apprise API over HTTP.
Questions or Feedback?
Technical Issues
Having trouble with the code? Open an issue on GitHub: