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 db, Notification, model_to_dict, fn from ..dependencies import oauth2_scheme, valid_token, LOG_DATA logging.basicConfig( filename=LOG_DATA['filename'], format=LOG_DATA['format'], level=LOG_DATA['log_level'] ) 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): all_notif = Notification.select().order_by(Notification.id) if all_notif.count() == 0: db.close() raise HTTPException(status_code=404, detail=f'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) 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) db.close() return Response(content=return_val, media_type='text/csv') else: return_val = {'count': all_notif.count(), 'notifs': []} for x in all_notif: return_val['notifs'].append(model_to_dict(x)) db.close() 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 Exception: db.close() 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) db.close() return Response(content=return_val, media_type='text/csv') else: return_val = model_to_dict(this_notif) db.close() return return_val @router.post('') async def post_notif(notif: NotifModel, token: str = Depends(oauth2_scheme)): if not valid_token(token): logging.warning(f'Bad Token: {token}') db.close() 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) db.close() return return_val else: db.close() 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(f'Bad Token: {token}') db.close() 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 Exception: db.close() 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) db.close() 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(f'Bad Token: {token}') db.close() 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 Exception: db.close() raise HTTPException(status_code=404, detail=f'No notification found with id {notif_id}') count = this_notif.delete_instance() db.close() 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')