Metadata-Version: 2.4
Name: argus-api-client
Version: 0.8.0
Summary: A Python API client library for the Argus alert aggregator server
Author-email: Sikt - Kunnskapssektorens Tjenesteleverandør <kontakt@sikt.no>
License-Expression: Apache-2.0
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.9
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: simple_rest_client
Requires-Dist: iso8601
Dynamic: license-file

# Argus API Client
[![test badge](https://img.shields.io/github/actions/workflow/status/Uninett/pyargus/tox.yml?branch=master)](https://github.com/Uninett/pyargus/actions)
[![Ruff](https://img.shields.io/endpoint?url=https://raw.githubusercontent.com/astral-sh/ruff/main/assets/badge/v2.json)](https://github.com/astral-sh/ruff)

This is the official Python client library for the
[Argus](https://github.com/Uninett/Argus) API server.

The Argus server is an incident registry, capable of aggregating alerts from
multiple source systems. Argus also can send event notifications (via e-mail,
SMS, etc.) when incidents are created or resolved.

## Usage examples

The pyargus library models [the official API endoints of
Argus](https://argus-server.readthedocs.io/en/latest/api.html) as methods on an
API client object.

At the moment, only the methods and models needed to interact with
incident-related endpoints are supported.

The `Client` class is found in `pyargus.client`, and the various supported data
models, such as `Incident`, `Event`, `Acknowledgement` and `SourceSystem`, are
implemented in `pyargus.models`.

**Note:** Always pass timezone-aware `datetime` objects (e.g. `datetime.now(tz=timezone.utc)`) to the client. pyargus serializes datetimes verbatim, so a naive value reaches Argus without a UTC offset and gets recorded at the wrong time.

### Listing open incidents that have not been acknowledged

```pycon
>>> from pyargus.client import Client
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> for incident in c.get_incidents(open=True, acked=False):
...    print(incident)
...
Incident(pk=4, start_time=datetime.datetime(2021, 4, 4, 16, 37, 43, 293726, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), source=SourceSystem(pk=2, name='testnav', type='nav', user=3, base_url='http://localhost/'), source_incident_id='202430', details_url='http://localhost/search/event/202430', description='uninett-gsw2 BGP session with 158.38.3.112 is DOWN', level=5, ticket_url='', tags=MultiValueDict([('location', 'Teknobyen Innovasjonssenter'), ('kundetjeneste', 'Nett_CNaaS'), ('kunde', 'example.org'), ('event_type', 'bgpState'), ('alert_type', 'bgpDown'), ('room', '100'), ('organization', 'uninett.srv'), ('host', 'uninett-gsw2.uninett.no')]), stateful=True, open=True, acked=False)
Incident(pk=3, start_time=datetime.datetime(2021, 4, 4, 16, 32, 53, 128780, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), source=SourceSystem(pk=2, name='testnav', type='nav', user=3, base_url='http://localhost/'), source_incident_id='202429', details_url='http://localhost/search/event/202429', description='uninett-gsw1 BGP session with 158.38.3.112 is DOWN', level=5, ticket_url='', tags=MultiValueDict([('location', 'Teknobyen Innovasjonssenter'), ('kundetjeneste', 'Nett_CNaaS'), ('kunde', 'example.org'), ('event_type', 'bgpState'), ('alert_type', 'bgpDown'), ('host', 'uninett-gsw1.uninett.no'), ('room', '100'), ('organization', 'uninett.srv')]), stateful=True, open=True, acked=False)
Incident(pk=2, start_time=datetime.datetime(2017, 8, 31, 14, 58, 31, 118794, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), source=SourceSystem(pk=2, name='testnav', type='nav', user=3, base_url='http://localhost/'), source_incident_id='184296', details_url='http://localhost/search/event/184296', description='Link DOWN on Gi0/3 at oldsmobile.lab (Simple is better than complex)', level=5, ticket_url='', tags=MultiValueDict([('room', '113'), ('location', 'Teknobyen Innovasjonssenter'), ('organization', 'uninett.testlab'), ('kundetjeneste', 'Nett_CNaaS'), ('kunde', 'example.org'), ('event_type', 'linkState'), ('alert_type', 'linkDown'), ('host', 'oldsmobile.lab.uninett.no'), ('interface', 'Gi0/3')]), stateful=True, open=True, acked=False)
```

As you can see, the arguments given to `get_incidents()` are translated
verbatim into the arguments supported by the `/incidents` endpoint in the API.

### List only "my" incidents

The incidents API also has an `/incidents/mine` endpoint, which works just like
the `/incidents` endpoint, but searches only the incidents that were posted
by the connecting user. This is useful for glue services, when they need to
compare the list of open Argus incidents it has produced with the current list
of active alerts in its source system.

Example:

```pycon
>>> from pyargus.client import Client
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> for incident in c.get_my_incidents(open=True, acked=False):
...    print(incident)
...
Incident(pk=3, start_time=datetime.datetime(2021, 4, 4, 16, 32, 53, 128780, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=datetime.datetime(9999, 12, 31, 23, 59, 59, 999999), source=SourceSystem(pk=3, name='foobar, type='nav', user=4, base_url='http://localhost/'), source_incident_id='2716057', details_url='http://localhost/search/event/2716057', description='uninett-gsw1 BGP session with 158.38.3.112 is DOWN', level=5, ticket_url='', tags=MultiValueDict([('location', 'Teknobyen Innovasjonssenter'), ('kundetjeneste', 'Nett_CNaaS'), ('kunde', 'example.org'), ('event_type', 'bgpState'), ('alert_type', 'bgpDown'), ('host', 'uninett-gsw1.uninett.no'), ('room', '100'), ('organization', 'uninett.srv')]), stateful=True, open=True, acked=False)
```

### Post a new incident

```pycon
>>> from pyargus.client import Client
>>> from pyargus.models import Incident
>>> from pyargus.time import now as utcnow
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> i = Incident(
...     description="The earth was demolished to make way for a hyperspace bypass",
...     start_time=utcnow(),
...     tags={
...         "host": "earth.example.org",
...     }
... )
>>> c.post_incident(i)
Incident(pk=8, start_time=datetime.datetime(2021, 4, 22, 11, 41, 53, 580947, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=None, source=SourceSystem(pk=2, name='testnav', type='nav', user=3, base_url='http://localhost/'), source_incident_id='', details_url='', description='The earth was demolished to make way for a hyperspace bypass', level=5, ticket_url='', tags=MultiValueDict([('host', 'earth.example.org')]), stateful=False, open=False, acked=False)
```

The `post_incident()` method returns the full `Incident` record, as stored in
Argus. If you need it, you can get the incident ID from the the primary key
attribute `pk`, in case you need to address it directly later.

### Close an existing incident

Incidents are closed by posting a *END* type event to an incident's event
log, with an optional timestamp. The `Client` class provides the follow
convenience method for this operation:

```pycon
>>> from pyargus.client import Client
>>> from pyargus.time import now as utcnow
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> c.resolve_incident(incident=8, description="The demolition was cancelled", timestamp=utcnow())
Event(pk=10, actor='testnav', description='The demolition was cancelled', incident=8, received=datetime.datetime(2021, 4, 22, 11, 47, 11, 978438, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), timestamp=datetime.datetime(2021, 4, 22, 11, 47, 11, 946076, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), type='END')
```

### Restart an existing incident

Incidents are restarted by posting a *RES* type event to an incident's event
log, with an optional timestamp. The `Client` class provides the follow
convenience method for this operation:

```pycon
>>> from pyargus.client import Client
>>> from pyargus.time import now as utcnow
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> c.restart_incident(incident=8, description="The demolition was restarted", timestamp=utcnow())
Event(pk=10, actor='testnav', description='The demolition was restarted', incident=8, received=datetime.datetime(2021, 4, 22, 11, 47, 11, 978438, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), timestamp=datetime.datetime(2021, 4, 22, 11, 47, 11, 946076, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), type='RES')
```

### Modify an existing incident

Argus does not allow modification of most incident attributes, but things
like the tag list can be changed. Modifications are made by constructing an
`Incident` object with the `pk` attribute set to the id of the incident you
wish you modify, and then adding values to the attributes you wish to modify:

```pycon
>>> from pyargus.client import Client
>>> from pyargus.models import Incident
>>> from datetime import datetime
>>> c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> i = Incident(
...     pk=8,
...     tags={
...         "host": "earth.example.org",
...         "location": "Milky way",
...     }
... )
>>> c.update_incident(i)
Incident(pk=8, start_time=datetime.datetime(2021, 4, 22, 11, 41, 53, 580947, tzinfo=datetime.timezone(datetime.timedelta(seconds=7200), '+02:00')), end_time=None, source=SourceSystem(pk=2, name='testnav', type='nav', user=3, base_url='http://localhost/'), source_incident_id='', details_url='', description='The earth was demolished to make way for a hyperspace bypass', level=None, ticket_url='', tags=MultiValueDict([('host', 'earth.example.org'), ('location', 'Milky way')]), stateful=False, open=False, acked=False)

```

### Working with multi-valued tags

Argus allows an incident to carry the same tag key more than once (for example
two different `host` values). The `tags` attribute is therefore a
`MultiValueDict` — a `dict` subclass that keeps every value while still behaving
like an ordinary dictionary for code that expects a single value per key.

Construct it from a sequence of `(key, value)` pairs (unlike a plain `dict`,
repeated keys are kept), or build it up with `add()`:

```pycon
>>> from pyargus.models import Incident
>>> from pyargus.multivaluedict import MultiValueDict
>>> tags = MultiValueDict([("host", "a.example.org"), ("host", "b.example.org")])
>>> tags.add("location", "Milky way")
>>> i = Incident(description="Two hosts affected", tags=tags)
```

Reading a multi-valued key through the ordinary `[]`/`get()` interface returns
only the *last* value and emits a `DeprecationWarning`, because the other values
are dropped silently. Use `getlist()` to retrieve them all:

```pycon
>>> tags.getlist("host")          # every value for the key
['a.example.org', 'b.example.org']
>>> tags["host"]                  # last value only; warns that values are dropped
'b.example.org'
>>> tags["location"]              # single-valued keys read normally, no warning
'Milky way'
>>> tags.lists()                  # all keys with all of their values
[('host', ['a.example.org', 'b.example.org']), ('location', ['Milky way'])]
>>> tags.allitems()               # every (key, value) pair, including repeats
[('host', 'a.example.org'), ('host', 'b.example.org'), ('location', 'Milky way')]
```

Passing a plain `dict` as `tags` still works exactly as before, so existing code
needs no changes.

### Stateless incidents

Argus supports a concept of "stateless" incidents. Stateless incidents
represent single points in time, and do not have an end time. To explicitly
create stateless incidents, set the `end_time` attribute to the STATELESS
sentinel, like so:

```pycon
from pyargus.models import Incident, STATELESS
from pyargus.time import now as utcnow

stateless_incident = Incident(
    description="Something happened",
    start_time=utcnow(),
    end_time=STATELESS
)
```


### Get a new authentication token

If your argus server is running version 1.29.0 or newer you can request to get
a new token (with a new expiration date) via API version 2. The token you are
using to access the server with must still be valid.

```python
tokenobj = c.refresh_token()
c = Client(api_root_url="https://argus.example.org/api/v2", token=tokenobj.token)
# save the contents of tokenobj in an environment variable, config file or
# secrets file so that it is not lost on program exit
```

### Send a heartbeat

A source system with no incidents to report looks exactly like one that has
crashed. Every glue service should therefore send a *heartbeat* on a regular
schedule to prove it is still alive. A heartbeat updates the source system's
`last_seen` timestamp on the server without posting an incident, so Argus can
tell a healthy-but-quiet source apart from a dead one.

```python
c.send_heartbeat()
```

The method returns `None` on success. On failure it raises an exception from
`simple_rest_client`, the HTTP client library that pyargus is built on; these
exceptions are defined in its `simple_rest_client.exceptions` module (for
example, `AuthError` for a `401` or `403` response).

#### Detecting whether the server supports heartbeats

The heartbeat endpoint requires an Argus server new enough to provide it.
Older servers don't fail cleanly on a `POST` — they reject it with a `403` that
is indistinguishable from an authentication error — so don't infer support from
the `send_heartbeat()` outcome. Instead, probe up front with
`supports_heartbeat()`, which issues a `GET` and returns whether the endpoint is
present:

```python
if c.supports_heartbeat():
    c.send_heartbeat()
```

A long-running glue service can check once at startup and decide whether to
bother sending heartbeats at all:

```python
c = Client(api_root_url="https://argus.example.org/api/v2", token="foobar")
heartbeats_enabled = c.supports_heartbeat()
while running:
    do_work()
    if heartbeats_enabled:
        c.send_heartbeat()
    sleep(interval)
```

## Async usage

An `AsyncClient` is available for use in asyncio-based applications. It mirrors
the `Client` interface, but all methods are coroutines. Use `python -m asyncio`
to try these examples interactively:

```pycon
>>> from pyargus.async_client import AsyncClient
>>> from pyargus.models import Incident
>>> from pyargus.time import now as utcnow
>>> c = AsyncClient(api_root_url="https://argus.example.org/api/v2", token="foobar")
>>> async for incident in c.get_incidents(open=True, acked=False):
...    print(incident)
...
Incident(pk=4, ...)
>>> i = Incident(
...     description="The earth was demolished to make way for a hyperspace bypass",
...     start_time=utcnow(),
...     tags={"host": "earth.example.org"},
... )
>>> await c.post_incident(i)
Incident(pk=8, ...)
>>> if await c.supports_heartbeat():
...     await c.send_heartbeat()
...
```

## BUGS

* Doesn't provide high-level error handling yet.

## Development

### Code style

Pyargus uses *ruff* as a source code formatter. Ruff is part of the optional dev dependencies listed in
[pyproject.toml](./pyproject.toml)

A pre-commit hook will format new code automatically before committing.
To enable this pre-commit hook, run

```console
$ pre-commit install
```
