b852025088
- FastAPI application that monitors websites for updates - SQLite database with Website and WebsiteAlert models - REST API endpoints for website management and alerts - Background scheduler for automatic periodic checks - Content hashing to detect website changes - Health check endpoint and comprehensive documentation - Alembic migrations for database schema management - CORS middleware for cross-origin requests - Environment variable configuration support Co-Authored-By: Claude <noreply@anthropic.com>
27 lines
1.0 KiB
Python
27 lines
1.0 KiB
Python
from typing import List
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from app.db.session import get_db
|
|
from app.models.website import WebsiteAlert
|
|
from app.api.schemas import AlertResponse
|
|
|
|
router = APIRouter(prefix="/alerts", tags=["alerts"])
|
|
|
|
@router.get("/", response_model=List[AlertResponse])
|
|
def list_alerts(skip: int = 0, limit: int = 100, unread_only: bool = False, db: Session = Depends(get_db)):
|
|
query = db.query(WebsiteAlert)
|
|
if unread_only:
|
|
query = query.filter(not WebsiteAlert.is_read)
|
|
|
|
alerts = query.order_by(WebsiteAlert.detected_at.desc()).offset(skip).limit(limit).all()
|
|
return alerts
|
|
|
|
@router.put("/{alert_id}/mark-read")
|
|
def mark_alert_read(alert_id: int, db: Session = Depends(get_db)):
|
|
alert = db.query(WebsiteAlert).filter(WebsiteAlert.id == alert_id).first()
|
|
if alert is None:
|
|
raise HTTPException(status_code=404, detail="Alert not found")
|
|
|
|
alert.is_read = True
|
|
db.commit()
|
|
return {"message": "Alert marked as read"} |