File size: 2,983 Bytes
86e2833 50ac51e 86e2833 50ac51e 86e2833 | 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 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 | from datetime import date
from pathlib import Path
from typing import Optional
import sqlite3
import streamlit as st
from fastapi import FastAPI, Query
from fastapi.responses import RedirectResponse
# Chemin de la base de données
dbpath = Path(__file__).parent.parent / "database" / "db_logsv2.db"
# instanciation de l'API FastAPI
api = FastAPI(
title="SmartRescue API",
description="API to access SmartRescue events",
version="1.0.0",
)
@api.get("/", include_in_schema=False)
async def root():
"""
Route par défaut, qui redirige vers la page doc
"""
return RedirectResponse(url="/docs")
# Requête de base, si l'utilisateur de l'API
# ne spécifie pas de date de début ou de fin
base_query = """
SELECT log.id_log, log.timestamp, prompt.prompt, prompt.response,
status.status, origin.origin AS origin
FROM log
LEFT JOIN prompt ON log.id_prompt = prompt.id_prompt
LEFT JOIN status ON log.id_status = status.id_status
LEFT JOIN origin ON log.id_origin = origin.id_origin
"""
@api.get("/data")
async def get_data(
start_date: Optional[date] = Query(None, description="Date de début (YYYY-MM-DD)"),
end_date: Optional[date] = Query(None, description="Date de fin (YYYY-MM-DD)"),
):
"""
Retourne les événements de SmartRescue entre deux dates.
Args:
start_date (Optional[datetime.date]): Date de début (YYYY-MM-DD), optionnelle.
end_date (Optional[datetime.date]): Date de fin (YYYY-MM-DD), optionnelle.
Returns:
List[Dict[str, Any]]: Liste des enregistrements, avec chaque champ de la base de données
comme clé du dictionnaire.
"""
try:
# Connexion à la base de données
conn = sqlite3.connect(dbpath)
cursor = conn.cursor()
query = base_query
params = []
# Ajout des conditions de date si spécifiées
if start_date or end_date:
query += " WHERE "
conditions = []
if start_date:
conditions.append("date(log.timestamp) >= ?")
params.append(start_date.isoformat())
if end_date:
conditions.append("date(log.timestamp) <= ?")
params.append(end_date.isoformat())
query += " AND ".join(conditions)
query += " ORDER BY log.timestamp DESC"
cursor.execute(query, params)
rows = cursor.fetchall()
# Récupération des enregistrements
columns = [description[0] for description in cursor.description]
result = [dict(zip(columns, row)) for row in rows]
return result
except sqlite3.OperationalError as e:
return {"error": f"Database error: {str(e)}", "path": dbpath}
finally:
# pour cloturer proprement la connexion à la base de données
if "conn" in locals():
conn.close()
|