1# SPDX-FileCopyrightText: 2026 Defensive Lab Agency
2# SPDX-FileContributor: u039b <git@0x39b.fr>
3#
4# SPDX-License-Identifier: GPL-3.0-or-later
5
6import logging
7from datetime import datetime, timedelta
8from typing import Type
9
10from sqlalchemy import delete, func, select
11from sqlalchemy.orm import Session
12
13from mongoose.models import Base
14from mongoose.models.configuration import HistoryConfiguration
15
16logger = logging.getLogger(__name__)
17
18
[docs]
19class SqliteHistoryManager:
20 """Manages the size and duration of records in the SQLite database."""
21
[docs]
22 def __init__(self, session_factory, config: HistoryConfiguration):
23 """Initialize the SqliteHistoryManager.
24
25 Args:
26 session_factory: A sessionmaker instance.
27 config: History limiting configuration.
28 """
29 self.Session = session_factory
30 self.config = config
31
[docs]
32 def cleanup(self, table_class: Type[Base]):
33 """Perform cleanup for the specified table based on configuration.
34
35 Args:
36 table_class: The SQLAlchemy model class (table) to clean up.
37 """
38 if not self.config or not self.config.enable:
39 return
40
41 try:
42 if self.config.max_duration_days:
43 self._cleanup_by_duration(table_class)
44
45 if self.config.max_records:
46 self._cleanup_by_size(table_class)
47 except Exception as e:
48 logger.error(f"Error during database cleanup for {table_class.__tablename__}: {e}")
49
50 def _cleanup_by_duration(self, table_class: Type[Base]):
51 """Remove records older than the configured duration.
52
53 Args:
54 table_class: The SQLAlchemy model class (table) to clean up.
55 """
56 cutoff_date = datetime.now() - timedelta(days=self.config.max_duration_days)
57
58 with self.Session() as session:
59 # We assume the table has a 'time' column which is a DateTime
60 stmt = delete(table_class).where(table_class.time < cutoff_date)
61 result = session.execute(stmt)
62 session.commit()
63
64 if result.rowcount > 0:
65 logger.info(
66 f"Removed {result.rowcount} records from {table_class.__tablename__} "
67 f"older than {self.config.max_duration_days} days"
68 )
69
70 def _cleanup_by_size(self, table_class: Type[Base]):
71 """Remove older records if the table exceeds the maximum number of records.
72
73 Args:
74 table_class: The SQLAlchemy model class (table) to clean up.
75 """
76 with self.Session() as session:
77 count = session.query(func.count(table_class.id)).scalar()
78
79 if count > self.config.max_records:
80 to_delete = count - self.config.max_records
81
82 # Find the IDs of the oldest records to delete
83 # We order by time (or id if time is same) to remove older records first
84 oldest_ids_query = (
85 select(table_class.id).order_by(table_class.time.asc(), table_class.id.asc()).limit(to_delete)
86 )
87 oldest_ids = session.execute(oldest_ids_query).scalars().all()
88
89 if oldest_ids:
90 delete_stmt = delete(table_class).where(table_class.id.in_(oldest_ids))
91 session.execute(delete_stmt)
92 session.commit()
93 logger.info(
94 f"Removed {len(oldest_ids)} oldest records from {table_class.__tablename__} "
95 f"to respect max_records limit of {self.config.max_records}"
96 )