alerts: per-support-team webhook; printedparts routes low-stock to a chosen team

Support teams gain a webhookurl (migration 7d29 + API + settings-page field), so
a team is a notification target. send_webhook(url=) lets a caller override the
site default with a team's webhook. Printedparts gains a 'alert support team'
setting (printedparts_alert_supportteamid) + selector on its settings page;
low-stock alerts post to that team's webhook, falling back to the site
alert_webhook_url. Email leg unchanged. Same pattern extends to other alerting
plugins (printers low-toner next).
This commit is contained in:
cproudlock
2026-07-22 13:32:00 -04:00
parent 824863a95a
commit fb188fd302
9 changed files with 142 additions and 21 deletions

View File

@@ -64,6 +64,7 @@ def create_support_team():
team = SupportTeam(
teamname=data['teamname'],
teamurl=data.get('teamurl'),
webhookurl=data.get('webhookurl'),
isactive=data.get('isactive', True))
db.session.add(team)
@@ -99,7 +100,7 @@ def update_support_team(team_id: int):
f"Support team '{data['teamname']}' already exists",
http_code=409)
for key in ['teamname', 'teamurl', 'isactive']:
for key in ['teamname', 'teamurl', 'webhookurl', 'isactive']:
if key in data:
setattr(team, key, data[key])

View File

@@ -17,6 +17,9 @@ class SupportTeam(BaseModel):
supportteamid = db.Column(db.Integer, primary_key=True)
teamname = db.Column(db.String(100), unique=True, nullable=False)
teamurl = db.Column(db.Text) # ServiceNow group deep link, nullable
# Teams (or other) webhook this team receives alerts on; alerting plugins
# route to it by selecting this team. Nullable = no webhook for this team.
webhookurl = db.Column(db.Text)
# Contacts cascade-delete with the team.
contacts = db.relationship(

View File

@@ -225,16 +225,18 @@ def _webhook_payload(fmt, title, text):
}
def send_webhook(title, text):
"""POST an alert to the configured webhook (Teams, etc.). Best-effort:
returns (ok, error); a no-op ((False, None)) when no URL is configured, and
never raises so it can never block the caller."""
def send_webhook(title, text, url=None):
"""POST an alert to a webhook (Teams, etc.). `url` overrides the site
alert_webhook_url (e.g. a support team's own webhook); the payload format
still follows alert_webhook_format. Best-effort: returns (ok, error), a
no-op ((False, None)) when no URL resolves, and never raises."""
config = get_webhook_config()
if not config['url']:
target = (url or '').strip() or config['url']
if not target:
return False, None
try:
response = requests.post(
config['url'], json=_webhook_payload(config['format'], title, text),
target, json=_webhook_payload(config['format'], title, text),
timeout=10)
if response.status_code >= 400:
return False, f'HTTP {response.status_code}'