174 lines
8.5 KiB
Python
174 lines
8.5 KiB
Python
import psycopg2
|
|
import json
|
|
|
|
class DBManager:
|
|
def __init__(self, host, port, dbname, user, password):
|
|
self.conn = psycopg2.connect(
|
|
host=host, port=port, dbname=dbname, user=user, password=password
|
|
)
|
|
self.conn.autocommit = False
|
|
|
|
def close(self):
|
|
if self.conn:
|
|
self.conn.close()
|
|
|
|
def fetch_all(self, query, params=None):
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(query, params)
|
|
return cur.fetchall()
|
|
|
|
def execute(self, query, params=None):
|
|
with self.conn.cursor() as cur:
|
|
cur.execute(query, params)
|
|
self.conn.commit()
|
|
|
|
# --- Reference data loaders ---
|
|
def get_animateurs(self):
|
|
return self.fetch_all("SELECT code_animateur, animateur FROM animation.animateurs ORDER BY animateur")
|
|
|
|
def get_lieux(self):
|
|
return self.fetch_all("SELECT code_lieux, lieux FROM animation.lieux_anim ORDER BY lieux")
|
|
|
|
def get_themes(self):
|
|
return self.fetch_all("SELECT code_type_anim, theme_animation FROM animation.theme_animation ORDER BY theme_animation")
|
|
|
|
def get_types_groupe(self):
|
|
return self.fetch_all("SELECT code_groupe, type_groupe FROM animation.type_groupe ORDER BY type_groupe")
|
|
|
|
def get_codes_annulation(self):
|
|
return self.fetch_all("SELECT code_annul, type_annulation FROM animation.code_annulation ORDER BY type_annulation")
|
|
|
|
def get_type_dossier_ens(self):
|
|
return self.fetch_all("SELECT code_dossier_ens, type_dossier_ens FROM animation.type_dossier_ens ORDER BY type_dossier_ens")
|
|
|
|
def get_etablissements(self):
|
|
return self.fetch_all("SELECT code_etablissement, nom_etablissement, commune FROM animation.liste_etablissements ORDER BY nom_etablissement")
|
|
|
|
def get_all_animations(self):
|
|
"""Retourne toutes les animations pour la liste déroulante (date + lieu + thème)."""
|
|
return self.fetch_all("""
|
|
SELECT da.code_animation,
|
|
da.date_anim,
|
|
da.lieux,
|
|
ta.theme_animation
|
|
FROM animation.donnees_animation da
|
|
LEFT JOIN animation.theme_animation ta ON ta.code_type_anim = da.code_type_anim
|
|
ORDER BY da.date_anim DESC, da.code_animation DESC
|
|
""")
|
|
|
|
def get_animation_by_id(self, code_animation):
|
|
rows = self.fetch_all(
|
|
"SELECT * FROM animation.donnees_animation WHERE code_animation = %s", (code_animation,)
|
|
)
|
|
if rows:
|
|
with self.conn.cursor() as cur:
|
|
cur.execute("SELECT * FROM animation.donnees_animation WHERE code_animation = %s", (code_animation,))
|
|
cols = [d[0] for d in cur.description]
|
|
cur.execute("SELECT * FROM animation.donnees_animation WHERE code_animation = %s", (code_animation,))
|
|
row = cur.fetchone()
|
|
return dict(zip(cols, row)) if row else None
|
|
return None
|
|
|
|
# --- Insertions ---
|
|
def add_animateur(self, nom):
|
|
rows = self.fetch_all("SELECT MAX(code_animateur) FROM animation.animateurs")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
self.execute("INSERT INTO animation.animateurs (code_animateur, animateur) VALUES (%s, %s)", (next_code, nom))
|
|
return next_code, nom
|
|
|
|
def add_lieux(self, lieux, type_lieux='', type_lieux_regroupement='', geom_wkt=None, srid=4326):
|
|
rows = self.fetch_all("SELECT MAX(code_lieux) FROM animation.lieux_anim")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
if geom_wkt:
|
|
self.execute(
|
|
"INSERT INTO animation.lieux_anim (code_lieux, lieux, type_lieux, type_lieux_regroupement, geom) "
|
|
"VALUES (%s,%s,%s,%s,ST_SetSRID(ST_GeomFromText(%s),%s))",
|
|
(next_code, lieux, type_lieux, type_lieux_regroupement, geom_wkt, srid)
|
|
)
|
|
else:
|
|
self.execute(
|
|
"INSERT INTO animation.lieux_anim (code_lieux, lieux, type_lieux, type_lieux_regroupement) VALUES (%s,%s,%s,%s)",
|
|
(next_code, lieux, type_lieux, type_lieux_regroupement)
|
|
)
|
|
return next_code, lieux
|
|
|
|
def add_theme(self, theme, famille=''):
|
|
rows = self.fetch_all("SELECT MAX(code_type_anim) FROM animation.theme_animation")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
self.execute("INSERT INTO animation.theme_animation (code_type_anim, theme_animation, famille_theme_animation) VALUES (%s,%s,%s)",
|
|
(next_code, theme, famille))
|
|
return next_code, theme
|
|
|
|
def add_annulation(self, type_annulation):
|
|
rows = self.fetch_all("SELECT MAX(code_annul) FROM animation.code_annulation")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
self.execute("INSERT INTO animation.code_annulation (code_annul, type_annulation) VALUES (%s,%s)",
|
|
(next_code, type_annulation))
|
|
return next_code, type_annulation
|
|
|
|
def add_type_dossier(self, type_dossier):
|
|
rows = self.fetch_all("SELECT MAX(code_dossier_ens) FROM animation.type_dossier_ens")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
self.execute("INSERT INTO animation.type_dossier_ens (code_dossier_ens, type_dossier_ens) VALUES (%s,%s)",
|
|
(next_code, type_dossier))
|
|
return next_code, type_dossier
|
|
|
|
def add_etablissement(self, nom, commune, prive=False, geom_wkt=None, srid=4326):
|
|
rows = self.fetch_all("SELECT MAX(code_etablissement) FROM animation.liste_etablissements")
|
|
next_code = (rows[0][0] or 0) + 1
|
|
if geom_wkt:
|
|
self.execute(
|
|
"INSERT INTO animation.liste_etablissements (code_etablissement, nom_etablissement, prive, commune, geom) "
|
|
"VALUES (%s,%s,%s,%s,ST_SetSRID(ST_GeomFromText(%s),%s))",
|
|
(next_code, nom, prive, commune, geom_wkt, srid)
|
|
)
|
|
else:
|
|
self.execute(
|
|
"INSERT INTO animation.liste_etablissements (code_etablissement, nom_etablissement, prive, commune) VALUES (%s,%s,%s,%s)",
|
|
(next_code, nom, prive, commune)
|
|
)
|
|
return next_code, nom, commune
|
|
|
|
def get_next_code_animation(self):
|
|
rows = self.fetch_all("SELECT MAX(code_animation) FROM animation.donnees_animation")
|
|
return (rows[0][0] or 0) + 1
|
|
|
|
def insert_donnees_animation(self, data: dict):
|
|
sql = """
|
|
INSERT INTO animation.donnees_animation (
|
|
code_animation, date_anim, duree_anim, animateurs, prepa_deplacemt, temps_total,
|
|
code_groupe, precisions_type_de_groupe, commune_provenance, nom_etablissement,
|
|
nbre_pers_total, nbre_pers_enfants, nbre_pers_adultes,
|
|
lieux, precisions_lieux_anim, accueil_pin, theme_detaille,
|
|
code_type_anim, payant, type_annul, remarques,
|
|
type_dossier_ens, realise_par_tiers, detail_tiers
|
|
) VALUES (
|
|
%(code_animation)s, %(date_anim)s, %(duree_anim)s, %(animateurs)s, %(prepa_deplacemt)s, %(temps_total)s,
|
|
%(code_groupe)s, %(precisions_type_de_groupe)s, %(commune_provenance)s, %(nom_etablissement)s,
|
|
%(nbre_pers_total)s, %(nbre_pers_enfants)s, %(nbre_pers_adultes)s,
|
|
%(lieux)s, %(precisions_lieux_anim)s, %(accueil_pin)s, %(theme_detaille)s,
|
|
%(code_type_anim)s, %(payant)s, %(type_annul)s, %(remarques)s,
|
|
%(type_dossier_ens)s, %(realise_par_tiers)s, %(detail_tiers)s
|
|
)
|
|
"""
|
|
self.execute(sql, data)
|
|
|
|
def update_donnees_animation(self, code_animation, data: dict):
|
|
sql = """
|
|
UPDATE animation.donnees_animation SET
|
|
date_anim=%(date_anim)s, duree_anim=%(duree_anim)s, animateurs=%(animateurs)s,
|
|
prepa_deplacemt=%(prepa_deplacemt)s, temps_total=%(temps_total)s,
|
|
code_groupe=%(code_groupe)s, precisions_type_de_groupe=%(precisions_type_de_groupe)s,
|
|
commune_provenance=%(commune_provenance)s, nom_etablissement=%(nom_etablissement)s,
|
|
nbre_pers_total=%(nbre_pers_total)s, nbre_pers_enfants=%(nbre_pers_enfants)s,
|
|
nbre_pers_adultes=%(nbre_pers_adultes)s,
|
|
lieux=%(lieux)s, precisions_lieux_anim=%(precisions_lieux_anim)s,
|
|
accueil_pin=%(accueil_pin)s, theme_detaille=%(theme_detaille)s,
|
|
code_type_anim=%(code_type_anim)s, payant=%(payant)s, type_annul=%(type_annul)s,
|
|
remarques=%(remarques)s, type_dossier_ens=%(type_dossier_ens)s,
|
|
realise_par_tiers=%(realise_par_tiers)s, detail_tiers=%(detail_tiers)s
|
|
WHERE code_animation=%(code_animation)s
|
|
"""
|
|
data['code_animation'] = code_animation
|
|
self.execute(sql, data)
|