Python

One call when your app starts, plus your web framework's integration. The package has no dependencies and supports Python 3.9 and newer.

5 min read

Install

bash

pip install naijacloud-argus

Set up

When your app starts, as early as you can (the top of your entry file, settings.py, wsgi.py or asgi.py):

Set up

python

import naijacloud_argus as argus

argus.init(dsn="https://<public key>@argus-in.naijacloud.com/<project id>")

init() with no dsn reads ARGUS_DSN, then SENTRY_DSN. A Naijacloud service gets the DSN from its Send errors to Argus switch, so there argus.init() is enough.

Send a test event

bash

ARGUS_DSN=https://<public key>@argus-in.naijacloud.com/<project id> \
  python -c 'import naijacloud_argus as argus; argus.init(); argus.capture_message("Argus test event")'

The SDK sends what's queued when the process exits, so the event shows under Argus → Errors within a minute. If it doesn't, add debug=True to init() to print what the SDK does.

What it captures

With only init():

Uncaught exceptions, on the main thread and in other threads. Python still prints the traceback and exits as it would without Argus.
logging records as breadcrumbs on the next error. No handler is added, so your logging setup is untouched.
Outgoing HTTP through urllib, requests or urllib3, as breadcrumbs, and inside a span as child spans with a traceparent header.
A final flush at exit, so a short script's data isn't lost.

Web frameworks catch your errors themselves, so add their integration too.

Frameworks

python

# settings.py
import naijacloud_argus as argus
argus.init()

MIDDLEWARE = [
    "naijacloud_argus.integrations.django.ArgusMiddleware",  # first
    # ...
]

Each integration traces every request, named after its route (GET /orders/<int:pk>/), and reports unhandled errors. A traceparent header from the caller continues its trace, and each request gets its own scope, so set_user() in one request never leaks into another.

Django: works under WSGI and ASGI, with sync and async views.
Flask: errors your own @app.errorhandler handles are not reported.
FastAPI: HTTPExceptions are answers, not errors, and are not reported.

Report errors yourself

Errors

python

import naijacloud_argus as argus

try:
    charge_card(order)
except PaystackError as err:
    argus.capture_exception(err, tags={"provider": "paystack"}, extra={"order_id": order.id})

argus.capture_message("Webhook signature mismatch", "warning")
argus.set_user({"id": user.id})
argus.set_tag("region", "af-west")

Inside an except block, capture_exception() with no argument sends the exception being handled. with argus.new_scope(): keeps a user or tags set inside it from leaking out, for background jobs and queue workers.

Logs

python

argus.log.info("order placed", order_id="ord_81", total=18500)
argus.log.error("paystack timed out", attempt=3)

Logs are sent in batches every 5 seconds, and a log written inside a request carries its trace id. To ship your logging records too, use argus.init(capture_logs=True) (INFO and up) or pass a level, such as capture_logs=logging.WARNING.

Metrics

python

argus.metrics.count("orders.placed", 1, {"method": "card"})
argus.metrics.gauge("queue.depth", len(jobs))
argus.metrics.distribution("checkout.ms", elapsed, unit="ms")

Metrics are added up in your process and sent every 10 seconds.

Traces

python

with argus.trace("charge card", {"provider": "paystack"}) as span:
    span.set_attribute("amount", 18500)
    paystack.charge(order)

@argus.trace("nightly report")
async def build_report(): ...

A span joins the request or span it runs in, and is marked failed when the block raises. Sample traces with traces_sample_rate, from 0 to 1 (default 1).

Cron check-ins

Create a cron monitor in Argus first, then report runs by its slug:

Cron check-ins

python

@argus.cron.monitor("nightly-invoices")
def send_invoices(): ...

# or by hand:
argus.cron.check_in("nightly-invoices", "start")
argus.cron.check_in("nightly-invoices", "ok")   # or "error", or an exit code

monitor reports the start, then success or failure, and re-raises your job's error. It also works as a with block and on async functions.

Feature flags

python

if argus.flags.is_on("new-checkout", user.id):
    ...
theme = argus.flags.variant("theme", user.id, "light")

# in async code, so the event loop isn't blocked:
if await argus.flags.is_on_async("new-checkout", user.id):
    ...

Answers are cached for 30 seconds per user. If Argus doesn't answer within flags_timeout (2 seconds), or the flag doesn't exist, you get the default.

Release and environment

init() reads ARGUS_RELEASE and ARGUS_ENVIRONMENT, then SENTRY_RELEASE and SENTRY_ENVIRONMENT. The environment is production when none is set. Or pass them:

Release and environment

python

argus.init(release=os.environ.get("GIT_SHA"), environment="staging")

Privacy

On by default:

Values of keys such as password, token, secret, authorization, cookie, bvn and card become [Filtered], at any depth.
Card numbers, bearer tokens, JWTs and password=… pairs are filtered inside strings, including exception messages and log lines.
Query strings are removed from URLs in breadcrumbs, requests and spans.
Request bodies and cookies are never sent.
The user's IP address is not sent unless you set send_default_pii=True.

before_send(event, hint) changes an event, or drops it when it returns None. hint["exc_info"] is the original exception:

before_send

python

def before_send(event, hint):
    exc = hint.get("exc_info")
    if exc and isinstance(exc[1], ConnectionResetError):
        return None
    return event

argus.init(before_send=before_send)

Already on Sentry

Keep sentry-sdk. In your sentry_sdk.init call, change the dsn line:

sentry_sdk.init

python

dsn="https://<public key>@argus-in.naijacloud.com/<project id>",