from datetime import datetime from fastapi import APIRouter, Depends, HTTPException, Response from typing import Optional import logging import pydantic from pandas import DataFrame from ..db_engine import Notification, model_to_dict, fn, DoesNotExist from ..dependencies import oauth2_scheme, valid_token router = APIRouter(prefix="/api/v2/notifs", tags=["notifs"]) class NotifModel(pydantic.BaseModel): created: int title: str desc: Optional[str] = None field_name: str message: str about: Optional[str] = "blank" ack: Optional[bool] = False @router.get("") async def get_notifs( created_after: Optional[int] = None, title: Optional[str] = None, desc: Optional[str] = None, field_name: Optional[str] = None, in_desc: Optional[str] = None, about: Optional[str] = None, ack: Optional[bool] = None, csv: Optional[bool] = None, limit: Optional[int] = 100, ): if limit is not None: limit = max(0, min(limit, 500)) all_notif = Notification.select().order_by(Notification.id) if all_notif.count() == 0: raise HTTPException( status_code=404, detail="There are no notifications to filter" ) if created_after is not None: # Convert milliseconds timestamp to datetime for PostgreSQL comparison created_after_dt = datetime.fromtimestamp(created_after / 1000) all_notif = all_notif.where(Notification.created < created_after_dt) if title is not None: all_notif = all_notif.where(Notification.title == title) if desc is not None: all_notif = all_notif.where(Notification.desc == desc) if field_name is not None: all_notif = all_notif.where(Notification.field_name == field_name) if in_desc is not None: all_notif = all_notif.where( fn.Lower(Notification.desc).contains(in_desc.lower()) ) if about is not None: all_notif = all_notif.where(Notification.about == about) if ack is not None: all_notif = all_notif.where(Notification.ack == ack) total_count = all_notif.count() if limit is not None: all_notif = all_notif.limit(limit) if csv: data_list = [ ["id", "created", "title", "desc", "field_name", "message", "about", "ack"] ] for line in all_notif: data_list.append( [ line.id, line.created, line.title, line.desc, line.field_name, line.message, line.about, line.ack, ] ) return_val = DataFrame(data_list).to_csv(header=False, index=False) return Response(content=return_val, media_type="text/csv") else: return_val = {"count": total_count, "notifs": []} for x in all_notif: return_val["notifs"].append(model_to_dict(x)) return return_val @router.get("/{notif_id}") async def get_one_notif(notif_id, csv: Optional[bool] = None): try: this_notif = Notification.get_by_id(notif_id) except DoesNotExist: raise HTTPException( status_code=404, detail=f"No notification found with id {notif_id}" ) if csv: data_list = [ ["id", "created", "title", "desc", "field_name", "message", "about", "ack"], [ this_notif.id, this_notif.created, this_notif.title, this_notif.desc, this_notif.field_name, this_notif.message, this_notif.about, this_notif.ack, ], ] return_val = DataFrame(data_list).to_csv(header=False, index=False) return Response(content=return_val, media_type="text/csv") else: return_val = model_to_dict(this_notif) return return_val @router.post("") async def post_notif(notif: NotifModel, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning("Bad Token: [REDACTED]") raise HTTPException( status_code=401, detail="You are not authorized to post notifications. This event has been logged.", ) logging.info(f"new notif: {notif}") this_notif = Notification( created=datetime.fromtimestamp(notif.created / 1000), title=notif.title, desc=notif.desc, field_name=notif.field_name, message=notif.message, about=notif.about, ) saved = this_notif.save() if saved == 1: return_val = model_to_dict(this_notif) return return_val else: raise HTTPException( status_code=418, detail="Well slap my ass and call me a teapot; I could not save that notification", ) @router.patch("/{notif_id}") async def patch_notif( notif_id, created: Optional[int] = None, title: Optional[str] = None, desc: Optional[str] = None, field_name: Optional[str] = None, message: Optional[str] = None, about: Optional[str] = None, ack: Optional[bool] = None, token: str = Depends(oauth2_scheme), ): if not valid_token(token): logging.warning("Bad Token: [REDACTED]") raise HTTPException( status_code=401, detail="You are not authorized to patch notifications. This event has been logged.", ) try: this_notif = Notification.get_by_id(notif_id) except DoesNotExist: raise HTTPException( status_code=404, detail=f"No notification found with id {notif_id}" ) if title is not None: this_notif.title = title if desc is not None: this_notif.desc = desc if field_name is not None: this_notif.field_name = field_name if message is not None: this_notif.message = message if about is not None: this_notif.about = about if ack is not None: this_notif.ack = ack if created is not None: this_notif.created = datetime.fromtimestamp(created / 1000) if this_notif.save() == 1: return_val = model_to_dict(this_notif) return return_val else: raise HTTPException( status_code=418, detail="Well slap my ass and call me a teapot; I could not save that rarity", ) @router.delete("/{notif_id}") async def delete_notif(notif_id, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning("Bad Token: [REDACTED]") raise HTTPException( status_code=401, detail="You are not authorized to delete notifications. This event has been logged.", ) try: this_notif = Notification.get_by_id(notif_id) except DoesNotExist: raise HTTPException( status_code=404, detail=f"No notification found with id {notif_id}" ) count = this_notif.delete_instance() if count == 1: raise HTTPException( status_code=200, detail=f"Notification {notif_id} has been deleted" ) else: raise HTTPException( status_code=500, detail=f"Notification {notif_id} was not deleted" )