File size: 2,414 Bytes
dbb00c8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 | import os
import apprise
from fastapi import FastAPI, Request, HTTPException
from pydantic import BaseModel, Field, field_validator
from typing import List, Union
# --- Models for Request Body ---
class Notification(BaseModel):
urls: Union[List[str], str] = Field(..., description="A single Apprise URL or a list of them.")
body: str = Field(..., description="The main content of the notification.")
title: str = Field(None, description="The optional title of the notification.")
tag: str = Field(None, description="An optional tag to target pre-configured URLs.")
@field_validator('urls')
@classmethod
def urls_to_list(cls, v: Union[List[str], str]) -> List[str]:
if isinstance(v, str):
return [v]
return v
# --- FastAPI App Initialization ---
app = FastAPI(
title="Apprise API Wrapper",
description="A flexible wrapper for Apprise that is Hugging Face compatible.",
)
# --- Apprise Configuration Loading ---
CONFIG_FILE = 'apprise.yml'
apobj = apprise.Apprise()
if os.path.exists(CONFIG_FILE):
apobj.add(f"file://{CONFIG_FILE}")
# --- API Endpoints ---
@app.get("/")
def health_check():
"""
Health Check Endpoint.
"""
return {"status": "ok", "message": "Apprise API is healthy."}
@app.post("/notify")
async def notify(notification: Notification):
"""
Sends a notification using Apprise. Now accepts both string and list for 'urls'.
"""
request_apobj = apprise.Apprise()
for url in notification.urls:
request_apobj.add(url)
if notification.tag:
if not apobj.find(notification.tag):
raise HTTPException(
status_code=400,
detail=f"The specified tag '{notification.tag}' was not found in the configuration.",
)
request_apobj.add(apobj.find(notification.tag))
if not request_apobj.urls():
raise HTTPException(
status_code=400,
detail="No notification URLs provided in the request or matched by the tag.",
)
success = request_apobj.notify(
body=notification.body,
title=notification.title,
)
if not success:
raise HTTPException(
status_code=500,
detail="Failed to send notification. Check server logs for details.",
)
return {"status": "success", "message": "Notification sent successfully."} |