Pour interagir avec une base de données MySQL via une appilcation Flask, il est nécessaire d'utiliser SQLAlchemy pour la gestion ORM. Voici une configuration complète incluant les étapes de connexion, créasion de modèles, opérations CRUD et migrations de schéma.
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import text
from flask_migrate import Migrate
app = Flask(__name__)
# Paramètres de connexion centralisés
DB_HOST = "127.0.0.1"
DB_PORT = 3306
DB_USER = "flask"
DB_PASS = "CAPGfeskm8zbfcyh"
DB_NAME = "flask"
app.config['SQLALCHEMY_DATABASE_URI'] = (
f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
"?charset=utf8mb4"
)
db = SQLAlchemy(app)
migrate = Migrate(app, db)
# Validation de la connexion
with app.app_context():
try:
result = db.session.execute(text("SELECT 1"))
print(result.scalar())
except Exception as e:
print(f"Erreur de connexion : {e}")
Les modèles de données sont définis comme des classes héritant de db.Model. Les champs sont typés avec des contraintes spécifiques.
class Utilisateur(db.Model):
__tablename__ = "utilisateurs"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
nom = db.Column(db.String(100))
mot_de_passe = db.Column(db.String(100))
email = db.Column(db.String(100))
signature = db.Column(db.String(100))
class Article(db.Model):
__tablename__ = "articles"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
titre = db.Column(db.String(200), nullable=False)
contenu = db.Column(db.Text, nullable=False)
auteur_id = db.Column(db.Integer, db.ForeignKey("utilisateurs.id"))
auteur = db.relationship("Utilisateur", back_populates="articles")
Les opérations de base sont exposées via des routes Flask. Création d'un utilisateur :
@app.route("/creer")
def creer_utilisateur():
nouveau = Utilisateur(nom="Zhang San", mot_de_passe="111")
db.session.add(nouveau)
db.session.commit()
return "Création réussie"
Recherche de données avec filtre :
@app.route("/rechercher")
def rechercher_utilisateurs():
resultats = Utilisateur.query.filter(Utilisateur.nom == "Zhang San").all()
for item in resultats:
print(item.nom)
return "Recherche effectuée"
Mise à jour et suppression :
@app.route("/modifier")
def modifier_utilisateur():
cible = Utilisateur.query.filter_by(nom="Zhang San").first()
cible.mot_de_passe = "222"
db.session.commit()
return "Modification appliquée"
@app.route("/supprimer")
def supprimer_utilisateur():
utilisateur = Utilisateur.query.get(1)
db.session.delete(utilisateur)
db.session.commit()
return "Suppression confirmée"
Pour gérer les modifications de schéma, flask-migrate est utilisé. Après installation via pip, les commandes suivantes génèrent et appliquent les migrations :
flask db init
flask db migrate
flask db upgrade
Exemple complet incluant toutes les configurations :
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate
from sqlalchemy import text
app = Flask(__name__)
DB_HOST = "127.0.0.1"
DB_PORT = 3306
DB_USER = "flask"
DB_PASS = "CAPGfeskm8zbfcyh"
DB_NAME = "flask"
app.config['SQLALCHEMY_DATABASE_URI'] = (
f"mysql+pymysql://{DB_USER}:{DB_PASS}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
"?charset=utf8mb4"
)
db = SQLAlchemy(app)
migrate = Migrate(app, db)
class Utilisateur(db.Model):
__tablename__ = "utilisateurs"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
nom = db.Column(db.String(100))
mot_de_passe = db.Column(db.String(100))
email = db.Column(db.String(100))
signature = db.Column(db.String(100))
class Article(db.Model):
__tablename__ = "articles"
id = db.Column(db.Integer, primary_key=True, autoincrement=True)
titre = db.Column(db.String(200), nullable=False)
contenu = db.Column(db.Text, nullable=False)
auteur_id = db.Column(db.Integer, db.ForeignKey("utilisateurs.id"))
auteur = db.relationship("Utilisateur", back_populates="articles")
with app.app_context():
db.create_all()
@app.route("/")
def accueil():
return "Application Flask opérationnelle"
@app.route("/creer")
def creer_utilisateur():
nouveau = Utilisateur(nom="Zhang San", mot_de_passe="111")
db.session.add(nouveau)
db.session.commit()
return "Création réussie"
@app.route("/rechercher")
def rechercher_utilisateurs():
resultats = Utilisateur.query.filter(Utilisateur.nom == "Zhang San").all()
for item in resultats:
print(item.nom)
return "Recherche effectuée"
@app.route("/modifier")
def modifier_utilisateur():
cible = Utilisateur.query.filter_by(nom="Zhang San").first()
cible.mot_de_passe = "222"
db.session.commit()
return "Modification appliquée"
@app.route("/supprimer")
def supprimer_utilisateur():
utilisateur = Utilisateur.query.get(1)
db.session.delete(utilisateur)
db.session.commit()
return "Suppression confirmée"
if __name__ == "__main__":
app.run(debug=True)