- Introduction à Flask-APScheduler ===================================
Flask-APScheduler est une extension pour le framework Flask qui intègre la bibliothèque de planification de tâches APScheduler. Elle offre une solution robuste pour gérer les tâches en arrière-plan et les exécutions programmées au sein des applications Flask.
Cette extension permet de configurer facilement les planificateurs et les tâches via le système de configuration de Flask. Elle propose également une API RESTful intégrée pour la surveillance et la gestion des tâches, avec des fonctionnalités d'authentification et une liste blanche d'hôtes pour la sécurité. Son intégration native avec les Blueprints de Flask en fait un choix flexible pour les projets de toute taille.
- Installation ===============
Pour intégrer Flask-APScheduler dans votre projet, utilisez pip :
pip install Flask-APScheduler
- Utilisation Fondamentale ===========================
Voici un exemple de base montrant comment configurer et utiliser Flask-APScheduler avec différents types de déclencheurs.
import datetime
from flask import Flask
from flask_apscheduler import APScheduler
# Classe de configuration pour l'application Flask
class AppConfig:
SCHEDULER_API_ENABLED = True # Active l'API RESTful intégrée
# WERKZEUG_RUN_MAIN = True # Souvent nécessaire pour éviter des avertissements avec le reloader
app = Flask(__name__)
app.config.from_object(AppConfig())
# Initialisation du planificateur
planificateur = APScheduler()
planificateur.init_app(app)
planificateur.start()
# Déclencheur de type 'interval' : exécute la tâche toutes les 30 secondes
@planificateur.task('interval', id='mon_interval_job', seconds=30, misfire_grace_time=900)
def tache_periodique():
print(f"Tâche périodique exécutée à {datetime.datetime.now()}")
# Déclencheur de type 'cron' : exécute la tâche toutes les minutes
@planificateur.task('cron', id='ma_tache_quotidienne', minute='*')
def tache_quotidienne():
print(f"Tâche quotidienne exécutée à {datetime.datetime.now()}")
# Fonction pour une tâche avec déclencheur de type 'date'
def tache_specifique_date():
print(f"Tâche spécifique à une date exécutée à {datetime.datetime.now()}")
if __name__ == '__main__':
# Ajout d'une tâche avec un déclencheur de type 'date' pour une exécution unique
# Définir une date future pour que la tâche s'exécute lors du test
date_future = (datetime.datetime.now() + datetime.timedelta(minutes=1)).strftime('%Y-%m-%d %H:%M:%S')
planificateur.add_job(
id='mon_unique_job',
func=tache_specifique_date,
trigger='date',
run_date=date_future,
timezone="Europe/Paris",
replace_existing=True
)
app.run(use_reloader=False) # Important pour éviter les problèmes de double démarrage du planificateur
- Composants Clés d'APScheduler ================================
APScheduler est structuré autour de quatre composants principaux :
- Triggers (Déclencheurs) : Définissent quand une tâche doit s'exécuter.
- Job Stores (Magasins de Tâches) : Stockent les informations concernant les tâches.
- Executors (Exécuteurs) : Gèrent la manière dont les tâches sont exécutées (par exemple, dans un thread ou un processus séparé).
- Scheduler (Planificateur) : L'orchestrateur principal qui coordonne les autres composants.
4.1. Déclencheurs (Triggers)
Les déclencheurs définissent la logique de planification d'une tâche. Chaque tâche doit être associée à un déclencheur. Plusieurs types sont disponibles :
Déclencheur 'date'
Exécute une tâche une seule fois à une date et heure spécifiées.
run_date: Date et heure d'exécution (chaîne ou objet datetime).timezone: Fuseau horaire.
Déclencheur 'interval'
Exécute une tâche à intervalles réguliers.
weeks,days,hours,minutes,seconds: Durée de l'intervalle.start_date,end_date: Période d'activité du déclencheur.timezone: Fuseau horaire.
Déclencheur 'cron'
Permet une planification complexe basée sur des champs temporels similaires à cron.
year,month,day,week,day_of_week,hour,minute,second: Champs temporels.- Les valeurs peuvent être des entiers, des chaînes comme '*/5' (toutes les 5 unités), '1-5' (de 1 à 5), ou '1,3,5' (à 1, 3 et 5).
start_date,end_date,timezone.
Déclencheurs Composés (AndTrigger, OrTrigger)
Permettent de combiner plusieurs déclencheurs avec des opérateurs logiques. Nécessitent une importation explicite depuis apscheduler.triggers.combining.
AndTrigger(triggers: list, jitter: int | None): S'exécute si TOUS les déclencheurs sont actifs simultanément.OrTrigger(triggers: list, jitter: int | None): S'exécute si AU MOINS UN des déclencheurs est actif.jitter: Délai maximal de décalage aléatoire pour l'exécution (en secondes).
import datetime
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.combining import AndTrigger, OrTrigger
from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
from flask import Flask
from flask_apscheduler import APScheduler
# Déclencheur de date : exécution unique
date_declencheur = DateTrigger(run_date=(datetime.datetime.now() + datetime.timedelta(seconds=15)), timezone='Europe/Paris')
# Déclencheur d'intervalle : toutes les 5 secondes
interval_declencheur = IntervalTrigger(seconds=5, timezone='Europe/Paris')
# Déclencheur cron : chaque minute
cron_declencheur = CronTrigger(minute="*", second="0/10", timezone='Europe/Paris') # Chaque 10 secondes de chaque minute
# Déclencheur composé : AndTrigger (exemple conceptuel, peut nécessiter des ajustements pour fonctionner selon les besoins)
# Ici, la tâche s'exécutera seulement si les deux conditions (intervalle et jour de semaine) sont remplies
# Ce type de combinaison est généralement plus utile pour des cas précis ou des plages de temps chevauchantes
declencheur_compose_and = AndTrigger([
IntervalTrigger(minutes=1, timezone='Europe/Paris'),
CronTrigger(day_of_week="sat,sun", hour="9", minute="0", timezone='Europe/Paris') # Tous les samedis et dimanches à 9h00
])
# Déclencheur composé : OrTrigger
# La tâche s'exécutera soit le lundi à 2h, soit le mardi à 15h
declencheur_compose_or = OrTrigger([
CronTrigger(day_of_week='mon', hour='2', timezone='Europe/Paris'),
CronTrigger(day_of_week='tue', hour='15', timezone='Europe/Paris')
])
def executer_tache_multiple(type_declencheur):
print(f"Tâche du type '{type_declencheur}' exécutée à {datetime.datetime.now()}")
if __name__ == '__main__':
app_multi = Flask(__name__)
scheduler_multi = APScheduler(BackgroundScheduler(timezone='Europe/Paris'))
scheduler_multi.init_app(app_multi)
scheduler_multi.add_job(id='date_unique', func=executer_tache_multiple, trigger=date_declencheur, args=('date_unique', ), replace_existing=True)
scheduler_multi.add_job(id='interval_simple', func=executer_tache_multiple, trigger=interval_declencheur, args=('interval_simple', ), replace_existing=True)
scheduler_multi.add_job(id='cron_periodique', func=executer_tache_multiple, trigger=cron_declencheur, args=('cron_periodique', ), replace_existing=True)
# Note: Les AndTrigger peuvent être complexes à configurer pour des exécutions fréquentes.
# L'exemple ci-dessous est conçu pour montrer la syntaxe.
# scheduler_multi.add_job(id='compose_and', func=executer_tache_multiple, trigger=declencheur_compose_and, args=('compose_and', ), replace_existing=True)
scheduler_multi.add_job(id='compose_or', func=executer_tache_multiple, trigger=declencheur_compose_or, args=('compose_or', ), replace_existing=True)
scheduler_multi.start()
app_multi.run(port=5001, use_reloader=False)
4.2. Magasins de Tâches (Job Stores)
Les magasins de tâches définissent où les informations sur les tâches sont stockées. Par défaut, elles sont conservées en mémoire vive, mais APScheduler supporte diverses bases de données pour la persistance.
MemoryJobStore(par défaut) : Stockage en mémoire, non persistant.MongoDBJobStore: Pour les bases de données MongoDB.RedisJobStore: Pour Redis (stockage clé-valeur).SQLAlchemyJobStore: Supporte diverses bases de données relationnelles (SQLite, PostgreSQL, MySQL, etc.) via SQLAlchemy.
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.schedulers.background import BackgroundScheduler
# Configuration des magasins de tâches
config_magasins = {
'default': SQLAlchemyJobStore(url="sqlite:///jobs.sqlite"), # SQLite par défaut
# 'persistant_db': SQLAlchemyJobStore(url="mysql+pymysql://utilisateur:motdepasse@localhost:3306/ma_base_donnees"),
# 'cache_rapide': RedisJobStore(host="127.0.0.1", port=6379, db=0, password=''),
# 'doc_db': MongoDBJobStore(host="127.0.0.1", port=27017, database="taches_cron", collection="planifications")
}
# Le planificateur sera configuré avec ces magasins de tâches
# Exemple lors de la création du Scheduler ou via app.config pour Flask-APScheduler
planificateur_avec_stores = BackgroundScheduler(jobstores=config_magasins)
4.3. Exécuteurs (Executors)
Les exécuteurs sont responsables de l'exécution des tâches. Ils peuvent utiliser des threads ou des processus séparés pour exécuter les fonctions des tâches.
ThreadPoolExecutor(par défaut) : Exécute les tâches dans un pool de threads. Idéal pour les tâches I/O-bound.ProcessPoolExecutor: Exécute les tâches dans un pool de processus. Recommandé pour les tâches CPU-bound.
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
from apscheduler.schedulers.background import BackgroundScheduler
# Configuration des exécuteurs
config_executeurs = {
'default': ThreadPoolExecutor(20), # Pool de 20 threads par défaut
'processpool': ProcessPoolExecutor(5) # Pool de 5 processus
}
# Configuration des paramètres par défaut des tâches
parametres_taches_defaut = {
'coalesce': False, # Ne pas fusionner les exécutions manquées
'max_instances': 3, # Nombre maximum d'instances simultanées par tâche
'misfire_grace_time': 60 # Période de grâce pour les tâches manquées (60 secondes)
}
planificateur_avec_exec = BackgroundScheduler(executors=config_executeurs, job_defaults=parametres_taches_defaut)
4.4. Planificateur (Scheduler)
Le planificateur est le composant central qui gère l'ajout, la modification, la suppression et l'interrogation des tâches. APScheduler propose plusieurs types de planificateurs, adaptés à différents environnements d'application :
BlockingScheduler: Bloque le thread principal. Utile pour les scripts qui n'ont pas d'autre tâche à exécuter.BackgroundScheduler: Exécute les tâches en arrière-plan dans un thread séparé. Convient à la plupart des applications web.AsyncIOScheduler: Pour les applications basées surasyncio.GeventScheduler: Pour les applications utilisant Gevent.TornadoScheduler: Pour les applications web Tornado.TwistedScheduler: Pour les applications basées sur Twisted.QtScheduler: Pour les applications graphiques basées sur Qt.
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.mongodb import MongoDBJobStore
from apscheduler.jobstores.sqlalchemy import SQLAlchemyJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
# Exemple de configuration complète d'un planificateur (BackgroundScheduler)
# Méthode 1: Par passage direct d'arguments
job_stores_config = {
'mongo_store': MongoDBJobStore(host='localhost', port=27017, database='mon_app', collection='jobs'),
'sqlite_store': SQLAlchemyJobStore(url='sqlite:///mes_jobs.sqlite')
}
executors_config = {
'threadpool': ThreadPoolExecutor(max_workers=20),
'processpool': ProcessPoolExecutor(max_workers=5)
}
job_defaults_config = {
'coalesce': False,
'max_instances': 3,
'misfire_grace_time': 60
}
planificateur_complet_1 = BackgroundScheduler(
jobstores=job_stores_config,
executors=executors_config,
job_defaults=job_defaults_config,
timezone="Europe/Paris"
)
# Méthode 2: Par dictionnaire de configuration (compatible avec Flask-APScheduler)
config_dict_scheduler = {
'apscheduler.jobstores.mongo_store': {
'type': 'mongodb',
'host': 'localhost',
'port': '27017',
'database': 'mon_app',
'collection': 'jobs'
},
'apscheduler.jobstores.sqlite_store': {
'type': 'sqlalchemy',
'url': 'sqlite:///mes_jobs.sqlite'
},
'apscheduler.executors.threadpool': {
'class': 'apscheduler.executors.pool:ThreadPoolExecutor',
'max_workers': '20'
},
'apscheduler.executors.processpool': {
'type': 'processpool',
'max_workers': '5'
},
'apscheduler.job_defaults.coalesce': 'false',
'apscheduler.job_defaults.max_instances': '3',
'apscheduler.job_defaults.misfire_grace_time': '60',
'apscheduler.timezone': 'Europe/Paris',
}
planificateur_complet_2 = BackgroundScheduler(config_dict_scheduler)
# Méthode 3: Configuration après instanciation
planificateur_complet_3 = BackgroundScheduler()
planificateur_complet_3.configure(
jobstores=job_stores_config,
executors=executors_config,
job_defaults=job_defaults_config,
timezone='Europe/Paris'
)
- Configuration Spécifique à Flask-APScheduler ===============================================
Flask-APScheduler utilise des variables de configuration préfixées par SCHEDULER_ dans l'objet app.config de Flask.
5.1. Paramètres de Configuration Communs
SCHEDULER_API_ENABLED(bool) : Active/désactive l'API RESTful (par défaut :False).SCHEDULER_API_PREFIX(str) : Préfixe de l'URL pour l'API (par défaut : "/scheduler").SCHEDULER_ENDPOINT_PREFIX(str) : Préfixe pour les noms des endpoints Flask (par défaut : "scheduler.").SCHEDULER_ALLOWED_HOSTS(list) : Liste des hôtes autorisés à accéder à l'API (par défaut : ["*"]).SCHEDULER_JOBSTORES(dict) : Dictionnaire de configuration des magasins de tâches.SCHEDULER_EXECUTORS(dict) : Dictionnaire de configuration des exécuteurs.SCHEDULER_JOB_DEFAULTS(dict) : Dictionnaire des paramètres par défaut des tâches.SCHEDULER_TIMEZONE(str) : Fuseau horaire du planificateur.SCHEDULER_AUTH: Objet d'authentification pour l'API (ex:HTTPBasicAuth()).JOBS(list) : Liste des tâches à charger au démarrage de l'application (format dict).
# Exemple de configuration avancée dans une classe Config Flask
class ConfigAvancee:
SCHEDULER_API_ENABLED = True
SCHEDULER_API_PREFIX = '/api/planification'
SCHEDULER_ALLOWED_HOSTS = ["127.0.0.1", "localhost"] # Limite l'accès à l'API
SCHEDULER_JOBSTORES = {
'default': RedisJobStore(host='localhost', port=6379, db=1),
'backup': SQLAlchemyJobStore(url="postgresql://user:pass@host/mydb")
}
SCHEDULER_EXECUTORS = {
'default': ThreadPoolExecutor(max_workers=25),
'cpu_bound': ProcessPoolExecutor(max_workers=8)
}
SCHEDULER_JOB_DEFAULTS = {
'coalesce': True, # Fusionne les exécutions manquées
'max_instances': 2,
'misfire_grace_time': 120
}
SCHEDULER_TIMEZONE = 'Europe/Berlin'
# Définition des tâches directement via la configuration JOBS
JOBS = [
{
"id": "tache_initiale_1",
"func": "mon_app.tasks:print_message", # Référence à une fonction dans un module
"args": ["Bonjour du scheduler!"],
"trigger": "interval",
"minutes": 5
}
]
5.2. API RESTful Intégrée
Lorsque SCHEDULER_API_ENABLED est à True, Flask-APScheduler expose des endpoints RESTful pour interagir avec le planificateur.
GET /scheduler: Informations de base sur le planificateur.GET /scheduler/jobs: Liste de toutes les tâches.GET /scheduler/jobs/<job_id>: Détails d'une tâche spécifique.POST /scheduler/jobs(JSON) : Ajoute une nouvelle tâche.PATCH /scheduler/jobs/<job_id>(JSON) : Met à jour une tâche existante.DELETE /scheduler/jobs/<job_id>: Supprime une tâche.POST /scheduler/jobs/<job_id>/pause: Suspend une tâche.POST /scheduler/jobs/<job_id>/resume: Reprend une tâche.POST /scheduler/jobs/<job_id>/run: Exécute une tâche immédiatement.
5.3. Méthodes Courantes du Planificateur
L'objet scheduler offre une interface riche pour la gestion programmatique des tâches :
scheduler.start(): Démarre le planificateur.scheduler.shutdown(wait=True): Arrête le planificateur, attendant la fin des tâches par défaut.scheduler.pause(): Suspend toutes les exécutions futures.scheduler.resume(): Reprend les exécutions suspendues.scheduler.add_listener(callback, event_mask): Ajoute un écouteur d'événements (par ex.,EVENT_JOB_EXECUTED,EVENT_JOB_ERROR).scheduler.remove_listener(callback): Supprime un écouteur.scheduler.add_job(id, func, trigger, **kwargs): Ajoute une tâche.scheduler.remove_job(id, jobstore='default'): Suprime une tâche.scheduler.remove_all_jobs(jobstore='default'): Supprime toutes les tâches d'un magasin.scheduler.get_job(id, jobstore='default'): Récupère une tâche par son ID.scheduler.get_jobs(jobstore='default'): Récupère toutes les tâches.scheduler.modify_job(id, jobstore='default', **kwargs): Modifie les paramètres d'une tâche.scheduler.pause_job(id, jobstore='default'): Met en pause une tâche spécifique.scheduler.resume_job(id, jobstore='default'): Reprend une tâche spécifique.scheduler.run_job(id, jobstore='default'): Exécute une tâche spécifique immédiatement.scheduler.authenticate(auth_function): Définit une fonction d'authentification pour l'API.
- Exemples Détaillés =====================
6.1. Exemple 1 : Configuration via une classe Flask et Authentification API
Cet exemple illustre la configuration des tâches via la classe Config de Flask, l'utilisation d'un magasin de tâches persistant, et l'activasion d'une authentification basique pour l'API.
import datetime
import time
from apscheduler.jobstores.mongodb import MongoDBJobStore
from flask import Flask
from flask_apscheduler import APScheduler
from flask_apscheduler.auth import HTTPBasicAuth
# Fichier: main_app.py
class ProjetConfig:
# Définition des tâches à charger au démarrage
JOBS = [
{
"id": "calcul_periodique_id",
"func": "main_app:calcul_somme_periodique", # Référence à la fonction dans ce module
"args": [10, 20],
"trigger": "interval",
"seconds": 15,
'replace_existing': True # Remplace la tâche si elle existe déjà
},
{
"id": "affichage_cron_id",
'func': 'main_app:afficher_message_cron',
'args': ("message cron",),
"trigger": "cron",
'minute': "*",
"second": 30, # Exécute à la 30ème seconde de chaque minute
'replace_existing': True
}
]
# Magasin de tâches (ici, MongoDB)
SCHEDULER_JOBSTORES = {
'default': MongoDBJobStore(host="127.0.0.1", port=27017, database="planificateur_db", collection="taches")
}
# Exécuteurs
SCHEDULER_EXECUTORS = {"default": {"type": "threadpool", "max_workers": 20}}
# Paramètres par défaut des tâches
SCHEDULER_JOB_DEFAULTS = {"coalesce": False, "max_instances": 1}
# Authentification pour l'API du planificateur
SCHEDULER_AUTH = HTTPBasicAuth()
# Préfixe de l'API
SCHEDULER_API_PREFIX = '/gestion_taches'
# Activation de l'API
SCHEDULER_API_ENABLED = True
# Fuseau horaire
SCHEDULER_TIMEZONE = 'Europe/Paris'
# Hôtes autorisés pour l'API
SCHEDULER_ALLOWED_HOSTS = ["*"]
# Fonctions de tâche
def calcul_somme_periodique(val_a, val_b):
"""Fonction de démonstration : calcule la somme."""
time.sleep(2) # Simule un travail
somme = val_a + val_b
print(f"{datetime.datetime.now()} - Somme calculée: {somme}")
def afficher_message_cron(msg):
"""Affiche un message régulier."""
print(f"{datetime.datetime.now()} - Message CRON: {msg}")
if __name__ == "__main__":
app_instance = Flask(__name__)
app_instance.config.from_object(ProjetConfig())
planificateur_app = APScheduler()
planificateur_app.init_app(app_instance)
planificateur_app.start()
# Fonction d'authentification pour l'API
@planificateur_app.authenticate
def verifier_authentification(credentials):
"""Vérifie les identifiants pour l'accès à l'API du planificateur."""
return credentials["username"] == "admin" and credentials["password"] == "secure_pass"
app_instance.run(host='0.0.0.0', port=5002, use_reloader=False)
6.2. Exemple 2 : Tâches Concurrente avec Verrouillage de Processus
Cet exemple montre comment exécuter des tâches qui utilisent plusieurs processus (via multiprocessing) et inclut un mécanisme de verrouillage de fichier pour s'assurer que le planificateur ne démarre qu'une seule fois, même en mode de rechargement automatique de Flask (use_reloader=True).
import datetime
import time
import os
import atexit
import platform
from multiprocessing import Process
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.schedulers.background import BackgroundScheduler
from flask import Flask
from flask_apscheduler import APScheduler
# Ce fichier contient toute la logique pour un exemple autonome
# --- Fonctions de Tâches ---
def simuler_tache_longue(nom_tache):
"""Simule une tâche longue, comme un calcul intensif."""
time.sleep(5)
print(f"[{os.getpid()}] Tâche '{nom_tache}' terminée à {datetime.datetime.now()}")
def lancer_taches_paralleles():
"""Lance plusieurs instances de 'simuler_tache_longue' en parallèle."""
print(f"Début de 'lancer_taches_paralleles' à {datetime.datetime.now()}")
processus = []
for i in range(3):
p = Process(target=simuler_tache_longeur, args=(f"sub_tache_{i}",))
processus.append(p)
p.start()
for p in processus:
p.join() # Attendre la fin de tous les sous-processus
print(f"Fin de toutes les sous-tâches parallèles à {datetime.datetime.now()}")
# --- Configuration Flask-APScheduler ---
class SchedulerConfig:
JOBS = [
{
"id": "job_par_intervalles_process",
"func": "__main__:lancer_taches_paralleles", # Référence au module principal
"trigger": "interval",
"seconds": 10,
'replace_existing': True,
},
{
"id": "job_simple_process",
"func": "__main__:simuler_tache_longue",
"trigger": "interval",
"args": ("tache_directe_interval",),
"seconds": 5,
'replace_existing': True,
}
]
SCHEDULER_JOBSTORES = {
'default': RedisJobStore(host='localhost', port=6379, db=10)
}
SCHEDULER_EXECUTORS = {"default": {"type": "threadpool", "max_workers": 20}}
SCHEDULER_JOB_DEFAULTS = {"coalesce": True, "max_instances": 1}
SCHEDULER_API_PREFIX = '/planif_api'
SCHEDULER_API_ENABLED = True
SCHEDULER_TIMEZONE = 'Europe/Paris'
SCHEDULER_ALLOWED_HOSTS = ["*"]
# --- Initialisation de l'Application et du Planificateur ---
planificateur_glob = APScheduler(BackgroundScheduler(timezone='Europe/Paris'))
def creer_app():
app_inst = Flask(__name__)
app_inst.config.from_object(SchedulerConfig())
initialiser_planificateur(app_inst)
return app_inst
def initialiser_planificateur(app_flask):
"""
Assure que le planificateur ne démarre qu'une seule fois,
même avec le reloader de Flask. Utilise un verrou de fichier.
"""
if app_flask.debug and os.environ.get("WERKZEUG_RUN_MAIN") != "true":
# En mode debug, le reloader exécute le script deux fois.
# Nous ne voulons démarrer le planificateur qu'une seule fois.
return
# Verrou de fichier pour éviter le double démarrage du scheduler
lock_file_path = 'scheduler_instance.lock'
lock_file = None
try:
if platform.system() == 'Windows':
import msvcrt
lock_file = open(lock_file_path, 'wb')
msvcrt.locking(lock_file.fileno(), msvcrt.LK_NBLCK, 1)
else: # Systèmes Unix/Linux/macOS
import fcntl
lock_file = open(lock_file_path, 'wb')
fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
# Si le verrou a été acquis, nous pouvons initialiser et démarrer le planificateur
planificateur_glob.init_app(app_flask)
planificateur_glob.start()
print("Planificateur APScheduler démarré avec succès.")
# Enregistrer une fonction de déverrouillage à la sortie de l'application
def unlock_file_on_exit():
if lock_file:
try:
if platform.system() == 'Windows':
lock_file.seek(0)
msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1)
else:
fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
except Exception as e:
print(f"Erreur lors du déverrouillage du fichier : {e}")
finally:
lock_file.close()
# os.remove(lock_file_path) # Optionnel: supprimer le fichier de verrouillage
atexit.register(unlock_file_on_exit)
except Exception as e:
print(f"Impossible d'acquérir le verrou du planificateur ou erreur: {e}. Le planificateur ne sera pas démarré dans cette instance.")
if lock_file:
lock_file.close() # S'assurer de fermer le fichier même en cas d'échec d'acquisition du verrou
# --- Exécution principale ---
if __name__ == '__main__':
application = creer_app()
application.run(host='0.0.0.0', port=5003, debug=True, use_reloader=True)