mirror of
https://github.com/mealie-recipes/mealie.git
synced 2026-02-12 19:03:11 -05:00
68 lines
2.2 KiB
Python
68 lines
2.2 KiB
Python
from http.client import HTTPException
|
|
|
|
from fastapi import Depends, status
|
|
from sqlalchemy.orm.session import Session
|
|
|
|
from mealie.core.root_logger import get_logger
|
|
from mealie.db.db_setup import generate_session
|
|
from mealie.repos.all_repositories import get_repositories
|
|
from mealie.routes.routers import AdminAPIRouter
|
|
from mealie.schema.events import EventNotificationIn, EventNotificationOut, TestEvent
|
|
from mealie.services.events import test_notification
|
|
|
|
router = AdminAPIRouter()
|
|
|
|
logger = get_logger()
|
|
|
|
|
|
@router.post("/notifications")
|
|
async def create_event_notification(
|
|
event_data: EventNotificationIn,
|
|
session: Session = Depends(generate_session),
|
|
):
|
|
"""Create event_notification in the Database"""
|
|
db = get_repositories(session)
|
|
|
|
return db.event_notifications.create(event_data)
|
|
|
|
|
|
@router.post("/notifications/test")
|
|
async def test_notification_route(
|
|
test_data: TestEvent,
|
|
session: Session = Depends(generate_session),
|
|
):
|
|
"""Create event_notification in the Database"""
|
|
db = get_repositories(session)
|
|
|
|
if test_data.id:
|
|
event_obj: EventNotificationIn = db.event_notifications.get(test_data.id)
|
|
test_data.test_url = event_obj.notification_url
|
|
|
|
try:
|
|
test_notification(test_data.test_url)
|
|
except Exception as e:
|
|
logger.error(e)
|
|
raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR)
|
|
|
|
|
|
@router.get("/notifications", response_model=list[EventNotificationOut])
|
|
async def get_all_event_notification(session: Session = Depends(generate_session)):
|
|
"""Get all event_notification from the Database"""
|
|
db = get_repositories(session)
|
|
return db.event_notifications.get_all(override_schema=EventNotificationOut)
|
|
|
|
|
|
@router.put("/notifications/{id}")
|
|
async def update_event_notification(id: int, session: Session = Depends(generate_session)):
|
|
"""Update event_notification in the Database"""
|
|
# not yet implemented
|
|
raise HTTPException(status.HTTP_405_METHOD_NOT_ALLOWED)
|
|
|
|
|
|
@router.delete("/notifications/{id}")
|
|
async def delete_event_notification(id: int, session: Session = Depends(generate_session)):
|
|
"""Delete event_notification from the Database"""
|
|
# Delete Item
|
|
db = get_repositories(session)
|
|
return db.event_notifications.delete(id)
|