A bit of a beginner question: I'm trying to create a FastAPI route, that processes Prometheus alerts. The alert itself is just a payload that looks like this:
{
"endsAt": "0001-01-01T00:00:00Z",
"labels.prometheus": "label/prometheus-stack",
"labels.namespace": "namespace_here",
"labels.alertname": "JobFailed",
"labels.severity": "critical",
"labels.job_name": "uuid-job",
"labels.purpose": "purpose_here",
"labels.team": "team_here",
"status": "firing",
"startsAt": "2023-03-13T21:05:48.433Z",
"annotations.summary": "job failed",
"annotations.message": "Job XY failed.",
"fingerprint": "1234123412341234",
"generatorURL": "https://URL.stuff"
}
For the API I have the pydantic class "Alert" which looks like this:
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
annotations: Dict[str, str]
labels: Dict[str, str]
And all I want to do is to be able to post the Prometheus payload to the following FastAPI route and be able to access all properties and do something with it.
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
print(alert.fingerprint)
print(alert.annotations.summary)
print(alert.labels.alertname)
If I have the pydantic class like mentioned though, FastAPI answers with "422 Unprocessable Entity" since the fields "annotations" and "labels" are required. They themselves do not have any content, only the "child fields"(?) like labels.namespace, annotations.message, ... do have content.
If I edit the pydantic class to give these fields defaults and make it so they are not required anymore, like so:
annotations: Dict[str, str] = None
labels: Dict[str, str] = None
then the Post Request is successful, but the values are not desirable. The given Payload is given as
fingerprint='1234123412341234' status='firing' starts_at=datetime.datetime(2023, 3, 13, 21, 5, 48, 433000, tzinfo=datetime.timezone.utc) ends_at=datetime.datetime(1, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) generator_url='https://URL.stuff' annotations=None labels=None labels.purpose='purpose_here' labels.prometheus='label/prometheus-stack' labels.namespace='namespace_here' labels.severity='critical' annotations.summary='job failed' annotations.message='Job XY failed.' labels.alertname='JobFailed' labels.team='team_here' labels.job_name='uuid-job'
Note that "...annotations=None labels=None..." Also, the nested labels and annotations don't seem to be subscribeable in any way
print(alert.labels.namespace)
^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'NoneType' object has no attribute 'namespace'
I'm sure I just don't understand the handling of nested fields like this not nearly enough. But due to this I also do not know how to help myself and make this work. Any help is appreciated!
Kind regards
Your last comments shows some confusion about Python syntax. If your JSON data has a field named labels.prometheus, you can't ask your model for alert.labels.prometheus; that would suggest that your Alert object has a field named labels (that in turn has a field named prometheus).
You have a flat data structure. There is no hierarchy that would allow dotted expression like you're trying to use.
You'll need to create a model that maps those dotted field names (labels.prometheus) to field names that are compatible with Python syntax (e.g., labels_prometheus). We can do that using field aliases; for example:
from datetime import datetime
from pydantic import BaseModel, Field
from fastapi import FastAPI
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
labels_prometheus: str = Field(alias="labels.prometheus")
labels_namespace: str = Field(alias="labels.namespace")
labels_alertname: str = Field(alias="labels.alertname")
labels_severity: str = Field(alias="labels.severity")
labels_job_name: str = Field(alias="labels.job_name")
labels_purpose: str = Field(alias="labels.purpose")
labels_team: str = Field(alias="labels.team")
annotations_summary: str = Field(alias="annotations.summary")
annotations_message: str = Field(alias="annotations.message")
api = FastAPI()
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
If we deliver your example JSON document to this API, we will see printed on the console:
fingerprint='1234123412341234' status='firing'
startsAt=datetime.datetime(2023, 3, 13, 21, 5, 48, 433000,
tzinfo=datetime.timezone.utc) endsAt='0001-01-01T00:00:00Z'
labels_prometheus='label/prometheus-stack'
labels_namespace='namespace_here'
labels_alertname='JobFailed' labels_severity='critical'
labels_job_name='uuid-job' labels_purpose='purpose_here'
labels_team='team_here' annotations_summary='job failed'
annotations_message='Job XY failed.'
Read more in the Pydantic documentation.
Alternately, if you want a hiearchical data structure, you can transform the JSON input data via a root_validator. Something like this, maybe:
from datetime import datetime
from pydantic import BaseModel, Field, validator, root_validator
from fastapi import FastAPI
class Alert(BaseModel):
fingerprint: str
status: str
startsAt: datetime
endsAt: datetime
endsAt: str
labels: dict[str, str]
annotations: dict[str, str]
@root_validator(pre=True)
def split_dots(cls, values):
newvalues = {}
for k, v in values.items():
if "." in k:
head, tail = k.split(".")
newvalues.setdefault(head, {})[tail] = v
else:
newvalues[k] = v
return newvalues
api = FastAPI()
@api.post("/alerts/jobs", tags=["Alerts"])
async def recieve_failed_job_alert(alert: Alert):
print(alert)
This would print:
fingerprint='1234123412341234' status='firing'
startsAt=datetime.datetime(2023, 3, 13, 21, 5, 48,
433000, tzinfo=datetime.timezone.utc)
endsAt='0001-01-01T00:00:00Z' labels={'prometheus':
'label/prometheus-stack', 'namespace':
'namespace_here', 'alertname': 'JobFailed',
'severity': 'critical', 'job_name': 'uuid-job',
'purpose': 'purpose_here', 'team': 'team_here'}
annotations={'summary': 'job failed', 'message': 'Job
XY failed.'}