Daita Logo

from_db Monitors & Findings

Register local database monitors from an Agent.from_db() agent and inspect generated findings through agent.db.

#Overview

from_db() agents expose monitor and finding helpers through agent.db.

python
registered = agent.db.register_monitors([...])
print(agent.db.monitor_events)
print(agent.db.findings.open)

Monitors are local polling watches over the database. Findings are portable records of notable DB observations.

#Register Monitors

python
agent = await Agent.from_db(
    "postgresql://user:pass@host/db",
    mode="data_team",
)
 
agent.db.register_monitors([
    {
        "name": "Orders freshness",
        "type": "freshness",
        "severity": "warning",
        "entity": {"table": "orders"},
        "sql": "SELECT MAX(updated_at) FROM orders",
        "threshold": {"max_age_hours": 24},
        "interval": "1h",
    }
])

#Monitor Types

TypePurposeThreshold
freshnessAlert when a timestamp is too old{"max_age_hours": 24}
row_countAlert on low row counts or large changes{"min_rows": 1} or {"change_pct": 25}

#Findings

When a monitor triggers, from_db() records a finding:

python
for finding in agent.db.findings.open:
    print(finding["title"], finding["severity"])

Findings include:

  • id
  • title
  • severity
  • status
  • kind
  • source
  • entity
  • observed
  • evidence
  • timestamps

#Resolve Events

Monitor watches can also resolve findings when the condition clears. Resolved findings are available through:

python
resolved = agent.db.findings.resolved

#Custom Handler

Provide a handler when you want to send monitor events elsewhere:

python
async def notify(event, monitor):
    print("Monitor event:", monitor["name"], event.value)
 
agent.db.register_monitors(
    [
        {
            "name": "Products row count",
            "type": "row_count",
            "severity": "critical",
            "entity": {"table": "products"},
            "sql": "SELECT COUNT(*) FROM products",
            "threshold": {"min_rows": 1},
            "interval": "15m",
        }
    ],
    handler=notify,
)

#Export Findings

python
json_payload = agent.db.findings.export_json(indent=2)

Use this for local debugging, CI artifacts, or handoff to your own persistence layer.