107 lines
3.7 KiB
Python
107 lines
3.7 KiB
Python
"""Migration automatique SQLite — détecte et ajoute les colonnes manquantes."""
|
|
import logging
|
|
from sqlalchemy import text
|
|
from app.database import engine
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Définition des colonnes attendues par table : {table: [(col, type_sql, default)]}
|
|
EXPECTED_COLUMNS: dict[str, list[tuple[str, str, str | None]]] = {
|
|
"plant": [
|
|
("categorie", "TEXT", None),
|
|
("hauteur_cm", "INTEGER", None),
|
|
("maladies_courantes", "TEXT", None),
|
|
("astuces_culture", "TEXT", None),
|
|
("url_reference", "TEXT", None),
|
|
],
|
|
"garden": [
|
|
("latitude", "REAL", None),
|
|
("longitude", "REAL", None),
|
|
("altitude", "REAL", None),
|
|
("adresse", "TEXT", None),
|
|
("longueur_m", "REAL", None),
|
|
("largeur_m", "REAL", None),
|
|
("surface_m2", "REAL", None),
|
|
("carre_potager", "INTEGER", "0"),
|
|
("carre_x_cm", "INTEGER", None),
|
|
("carre_y_cm", "INTEGER", None),
|
|
("photo_parcelle", "TEXT", None),
|
|
("ensoleillement", "TEXT", None),
|
|
],
|
|
"task": [
|
|
("frequence_jours", "INTEGER", None),
|
|
("date_prochaine", "TEXT", None),
|
|
("outil_id", "INTEGER", None),
|
|
],
|
|
"meteostation": [
|
|
("t_min", "REAL", None),
|
|
("t_max", "REAL", None),
|
|
],
|
|
"tool": [
|
|
("photo_url", "TEXT", None),
|
|
("video_url", "TEXT", None),
|
|
("notice_fichier_url", "TEXT", None),
|
|
("boutique_nom", "TEXT", None),
|
|
("boutique_url", "TEXT", None),
|
|
("prix_achat", "REAL", None),
|
|
],
|
|
"planting": [
|
|
("boutique_nom", "TEXT", None),
|
|
("boutique_url", "TEXT", None),
|
|
("tarif_achat", "REAL", None),
|
|
("date_achat", "TEXT", None),
|
|
],
|
|
"plantvariety": [
|
|
# ancien nom de table → migration vers "plant" si présente
|
|
],
|
|
"media": [
|
|
("identified_species", "TEXT", None),
|
|
("identified_common", "TEXT", None),
|
|
("identified_confidence", "REAL", None),
|
|
("identified_source", "TEXT", None),
|
|
],
|
|
"astuce": [
|
|
("categorie", "TEXT", None),
|
|
("tags", "TEXT", None),
|
|
("mois", "TEXT", None),
|
|
("photos", "TEXT", None),
|
|
("videos", "TEXT", None),
|
|
],
|
|
}
|
|
|
|
|
|
def _existing_columns(conn, table: str) -> set[str]:
|
|
rows = conn.execute(text(f"PRAGMA table_info({table})")).fetchall()
|
|
return {row[1] for row in rows}
|
|
|
|
|
|
def _existing_tables(conn) -> set[str]:
|
|
rows = conn.execute(text("SELECT name FROM sqlite_master WHERE type='table'")).fetchall()
|
|
return {row[0] for row in rows}
|
|
|
|
|
|
def run_migrations() -> None:
|
|
with engine.connect() as conn:
|
|
tables = _existing_tables(conn)
|
|
|
|
# Renommage plantvariety → plant si l'ancienne table existe et la nouvelle non
|
|
if "plantvariety" in tables and "plant" not in tables:
|
|
conn.execute(text("ALTER TABLE plantvariety RENAME TO plant"))
|
|
conn.commit()
|
|
logger.info("Migration: plantvariety renommée en plant")
|
|
tables = _existing_tables(conn)
|
|
|
|
for table, columns in EXPECTED_COLUMNS.items():
|
|
if table not in tables or not columns:
|
|
continue
|
|
existing = _existing_columns(conn, table)
|
|
for col_name, col_type, default in columns:
|
|
if col_name not in existing:
|
|
if default is not None:
|
|
sql = f"ALTER TABLE {table} ADD COLUMN {col_name} {col_type} DEFAULT {default}"
|
|
else:
|
|
sql = f"ALTER TABLE {table} ADD COLUMN {col_name} {col_type}"
|
|
conn.execute(text(sql))
|
|
conn.commit()
|
|
logger.info(f"Migration: colonne {table}.{col_name} ajoutée")
|