Opérations de base de données Flask

Dans Flask, les opérations sur les bases de données sont un aspect important de la création d'applications web.

Flask offre plusieurs façons d'interagir avec les bases de données, notamment en utilisant directement SQL et en recourant à des outils ORM (mappage objet-relationnel) tels que SQLAlchemy.

Ce qui suit est une explication détaillée des opérations de base de données Flask, y compris les opérations de base utilisant SQLAlchemy et l'exécution directe d'instructions SQL.

  1. Utilisation de SQLAlchemy: Définir des modèles, configurer la base de données et effectuer des opérations CRUD de base.
  2. Création et gestion de bases de données: Utiliserdb.create_all()Créer une table.
  3. Opérations CRUD: Ajouter, lire, mettre à jour et supprimer des enregistrements.
  4. Opérations de requête: Exécuter des requêtes simples et complexes, y compris le tri et la pagination.
  5. Flask-Migrate: Utiliser Flask-Migrate pour gérer les migrations de base de données.
  6. Exécution de SQL brut: Utiliser des instructions SQL brutes pour les opérations de base de données.

1. Utilisation de SQLAlchemy

SQLAlchemy est une puissante bibliothèque ORM qui simplifie les opérations de base de données en interagissant avec les tables de base de données via des objets Python.

Flask-SQLAlchemy est une extension Flask permettant d'intégrer SQLAlchemy.

Installer Flask-SQLAlchemy

pip install flask-sqlalchemy

Configurer SQLAlchemy

Code du fichier app.py :

Exemple

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'  # Utiliser une base de données SQLite
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

2. Définir le modèle

Un modèle est une classe Python pour une table de base de données ; chaque classe de modèle représente une table dans la base de données.

Exemple

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

db.Model: toutes les classes de modèle doivent hériter de db.Model.

db.Column: définir les champs du modèle, en spécifiant des attributs tels que le type du champ, s'il s'agit d'une clé primaire, s'il est unique, s'il peut être nul, etc.

3. Création et gestion de bases de données

Créer la base de données et les tables

Après avoir défini les modèles, vous pouvez utiliser les méthodes fournies par SQLAlchemy pour créer la base de données et les tables.

with app.app_context():
    db.create_all()

db.create_all(): Crée les tables correspondant à tous les modèles définis dans le contexte actuel.

4. Opérations CRUD de base

Créer un enregistrement

Exemple

@app.route('/add_user')
def add_user():
    new_user = User(username='john_doe', email='[email protected]')
    db.session.add(new_user)
    db.session.commit()
    return 'Utilisateur ajouté !'

db.session.add(new_user): Ajoute le nouvel objet utilisateur à la session.

db.session.commit(): Valide la transaction pour enregistrer les modifications dans la base de données.

Lire un enregistrement

Exemple

@app.route('/get_users')
def get_users():
    users = User.query.all()  # Obtenir tous les utilisateurs
    return '<br>'.join([f'{user.username} ({user.email})' for user in users])

User.query.all(): Interroger tous les enregistrements utilisateur.

Mettre à jour l'enregistrement

Exemple

@app.route('/update_user/<int:user_id>')
def update_user(user_id):
    user = User.query.get(user_id)
    if user:
        user.username = 'new_username'
        db.session.commit()
        return 'Utilisateur mis à jour !'
    return 'Utilisateur non trouvé !'

User.query.get(user_id): Interroger un seul enregistrement utilisateur par clé primaire.

Mettre à jour les valeurs des champs et valider la transaction.

Supprimer l'enregistrement

Exemple

@app.route('/delete_user/<int:user_id>')
def delete_user(user_id):
    user = User.query.get(user_id)
    if user:
        db.session.delete(user)
        db.session.commit()
        return 'Utilisateur supprimé !'
    return 'Utilisateur non trouvé !'

db.session.delete(user): Supprimer l'enregistrement de l'utilisateur et valider la transaction.

5. Opérations de requête

SQLAlchemy offre une riche fonctionnalité de requête, vous permettant d'effectuer diverses opérations de requête via des objets de requête.

Requête de base

users = User.query.filter_by(username='john_doe').all()

filter_by(): Filtrer les enregistrements par valeur de champ.

Requête complexe

from sqlalchemy import or_

users = User.query.filter(or_(User.username == 'john_doe', User.email == '[email protected]')).all()

or_(): Utilisé pour exécuter des conditions de requête complexes.

Tri et pagination

users = User.query.order_by(User.username).paginate(page=1, per_page=10)

order_by(): Trier par le champ spécifié.

paginate(): Requête paginée.

6. Utiliser Flask-Migrate pour les migrations.

Flask-Migrate est une extension pour les migrations de bases de données, basée sur Alembic, qui vous aide à gérer le contrôle de version de la base de données.

Installer Flask-Migrate

pip install flask-migrate

Configurer Flask-Migrate

Code du fichier app.py :

from flask_migrate import Migrate

migrate = Migrate(app, db)

Initialiser la migration

flask db init

Créer des scripts de migration

flask db migrate -m "Initial migration."

Appliquer la migration

flask db upgrade
  • flask db init: Initialiser l'environnement de migration.
  • flask db migrate -m "message": Créer un script de migration.
  • flask db upgrade: Appliquer les migrations à la base de données.

7. Exécution de SQL brut

Bien que SQLAlchemy fournisse des fonctionnalités ORM, vous pouvez également exécuter des instructions SQL brutes.

Exemple

@app.route('/raw_sql')
def raw_sql():
    result = db.session.execute('SELECT * FROM user')
    return '<br>'.join([str(row) for row in result])

db.session.execute(): Exécute une requête SQL brute.

Connexion à une base de données MySQL et exploitation dans Flask

La connexion à une base de données MySQL et son exploitation dans Flask impliquent généralement d'utiliser SQLAlchemy ou directement les pilotes Python pour MySQL. Voici les étapes détaillées, incluant l'utilisation de Flask-SQLAlchemy et directement des pilotes Python pour MySQL.

1. Utiliser Flask-SQLAlchemy pour se connecter à MySQL

Flask-SQLAlchemy est une extension Flask qui simplifie la configuration et l'utilisation de SQLAlchemy. Pour vous connecter à MySQL, vous devez installer Flask-SQLAlchemy et un pilote MySQL.

Installez les bibliothèques requises :

pip install flask-sqlalchemy mysqlclient
  • flask-sqlalchemy: l'extension SQLAlchemy de Flask.
  • mysqlclient: pilote Python pour la base de données MySQL.

Configurer Flask-SQLAlchemy

Code du fichier app.py :

Exemple

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql://username:password@localhost/dbname'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

code>SQLALCHEMY_DATABASE_URI : Définit l'URI de connexion à la base de données, le format estmysql://username:password@localhost/dbname。

  • username: nom d'utilisateur MySQL.
  • password: mot de passe MySQL.
  • localhost: adresse de l'hôte MySQL (généralement local estlocalhost）。
  • dbname: nom de la base de données.

Définir les modèles et effectuer les opérations de base

Code du fichier app.py :

Exemple

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)

    def __repr__(self):
        return f'<User {self.username}>'

@app.route('/')
def index():
    users = User.query.all()
    return '<br>'.join([f'{user.username} ({user.email})' for user in users])

if __name__ == '__main__':
    with app.app_context():
        db.create_all()  # Créer la table de base de données
    app.run(debug=True)

Utilisez directement le pilote Python de MySQL.

Si vous choisissez de ne pas utiliser SQLAlchemy, mais d'utiliser directement le pilote Python de MySQL, vous pouvez utiliser la bibliothèque mysql-connector-python ou PyMySQL.

Installez mysql-connector-python:

pip install mysql-connector-python

Installez PyMySQL (si vous choisissez d'utiliser PyMySQL):

pip install PyMySQL

Utilisez mysql-connector-python pour vous connecter à MySQL

Code du fichier app.py :

Exemple

from flask import Flask, request, jsonify
import mysql.connector

app = Flask(__name__)

def get_db_connection():
    connection = mysql.connector.connect(
        host='localhost',
        user='username',
        password='password',
        database='dbname'
    )
    return connection

@app.route('/add_user', methods=['POST'])
def add_user():
    data = request.json
    name = data['name']
    email = data['email']

    connection = get_db_connection()
    cursor = connection.cursor()
    cursor.execute('INSERT INTO user (username, email) VALUES (%s, %s)', (name, email))
    connection.commit()
    cursor.close()
    connection.close()

    return 'Utilisateur ajouté !'

@app.route('/get_users')
def get_users():
    connection = get_db_connection()
    cursor = connection.cursor(dictionary=True)
    cursor.execute('SELECT * FROM user')
    users = cursor.fetchall()
    cursor.close()
    connection.close()

    return jsonify(users)

if __name__ == '__main__':
    app.run(debug=True)

Utilisez PyMySQL pour vous connecter à MySQL, code du fichier app.py :

Exemple

from flask import Flask, request, jsonify
import pymysql

app = Flask(__name__)

def get_db_connection():
    connection = pymysql.connect(
        host='localhost',
        user='username',
        password='password',
        database='dbname',
        cursorclass=pymysql.cursors.DictCursor
    )
    return connection

@app.route('/add_user', methods=['POST'])
def add_user():
    data = request.json
    name = data['name']
    email = data['email']

    connection = get_db_connection()
    with connection.cursor() as cursor:
        cursor.execute('INSERT INTO user (username, email) VALUES (%s, %s)', (name, email))
        connection.commit()

    connection.close()
    return 'Utilisateur ajouté !'

@app.route('/get_users')
def get_users():
    connection = get_db_connection()
    with connection.cursor() as cursor:
        cursor.execute('SELECT * FROM user')
        users = cursor.fetchall()

    connection.close()
    return jsonify(users)

if __name__ == '__main__':
    app.run(debug=True)
autres extensions