Database Manager

Le module DatabaseManager fournit une interface générique pour interagir avec une base de données SQLite.

Usage

Le DatabaseManager est utilisé par les modules de traitement, comme le DatabaseProcessor, pour stocker et récupérer les données en base.

Fonctionnalités

• Gère la connexion et la déconnexion à la base de données.

• Initialise le schéma de la base de données si nécessaire.

• Fournit des méthodes pour les opérations CRUD (Create, Read, Update, Delete).

• Permet l’exécution de requêtes SQL personnalisées.

Modules

class src.smart_watch.core.DatabaseManager.DatabaseManager(db_file: str | Path)

Bases : object

Gestionnaire de base de données avec SQLAlchemy.

__init__(db_file: str | Path)

Initialise le gestionnaire de base de données.

Paramètres:

db_file (Union[str, Path]) – chemin vers le fichier de base de données SQLite

get_session()

Retourne une nouvelle session.

close()

Ferme le moteur de base de données.

table_exists(table_name: str) → bool

Vérifie si une table existe dans la base de données.

Paramètres:

table_name (str) – nom de la table à vérifier

Renvoie:

True si la table existe, False sinon

Type renvoyé:

bool

initialize(table_name: str, df_initial: DataFrame, if_exists: str = 'fail') → None

Initialise une table dans la base de données SQLite avec les données de base.

Paramètres:

• table_name (str) – nom de la table à initialiser

• df_initial (pl.DataFrame) – dataFrame initial

• if_exists (str) – comportement si la table existe (« fail », « replace », « skip »)

Lève:

Exception – en cas d’erreur lors de l’initialisation

load_data(table_name: str, query: str | None = None) → DataFrame

Charge les données depuis une table de la base SQLite.

Paramètres:

• table_name (str) – nom de la table à charger

• query (Optional[str]) – requête SQL personnalisée

Renvoie:

dataFrame avec les données

Type renvoyé:

pl.DataFrame

Lève:

Exception – en cas d’erreur lors du chargement

update_record(table_name: str, where_conditions: Dict[str, Any], update_values: Dict[str, Any]) → int

Met à jour des enregistrements dans une table de la base SQLite.

Paramètres:

• table_name (str) – nom de la table à mettre à jour

• where_conditions (Dict[str, Any]) – conditions WHERE (colonne: valeur)

• update_values (Dict[str, Any]) – valeurs à mettre à jour (colonne: valeur)

Renvoie:

nombre d’enregistrements mis à jour

Type renvoyé:

int

Lève:

Exception – en cas d’erreur lors de la mise à jour

execute_query(query: str, params: tuple | None = None) → List[tuple]

Exécute une requête SQL personnalisée.

Paramètres:

• query (str) – requête SQL à exécuter

• params (Optional[tuple]) – paramètres pour la requête (optionnel)

Renvoie:

résultats de la requête

Type renvoyé:

List[tuple]