Operaciones de base de datos en Flask

En Flask, las operaciones de base de datos son un aspecto importante del desarrollo de aplicaciones web.

Flask ofrece varias formas de interactuar con la base de datos, incluyendo el uso directo de SQL y el uso de herramientas ORM (mapeo objeto-relacional), como SQLAlchemy.

A continuación se presenta una explicación detallada de las operaciones de base de datos en Flask, incluyendo las operaciones básicas con SQLAlchemy y la ejecución directa de sentencias SQL.

  1. Usar SQLAlchemy: definir modelos, configurar la base de datos y realizar operaciones CRUD básicas.
  2. Crear y gestionar la base de datos: usardb.create_all()para crear tablas.
  3. Operaciones CRUD: agregar, leer, actualizar y eliminar registros.
  4. Operaciones de consulta: realizar consultas básicas y complejas, incluyendo ordenación y paginación.
  5. Flask-Migrate: usar Flask-Migrate para gestionar migraciones de bases de datos.
  6. Ejecutar SQL puro: usar sentencias SQL puras para operaciones de base de datos.

1. Usar SQLAlchemy

SQLAlchemy es una potente biblioteca ORM que simplifica las operaciones de base de datos, permitiendo interactuar con las tablas mediante objetos de Python.

Flask-SQLAlchemy es una extensión de Flask que se utiliza para integrar SQLAlchemy.

Instalar Flask-SQLAlchemy

pip install flask-sqlalchemy

Configurar SQLAlchemy

Código del archivo app.py:

Ejemplo

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///example.db'  # Usar base de datos SQLite
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)

2. Definir modelos

Los modelos son clases de Python que representan tablas de la base de datos; cada clase de modelo representa una tabla.

Ejemplo

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: todas las clases de modelo deben heredar de db.Model.

db.Column: definir los campos del modelo, especificando el tipo de campo, si es clave primaria, si es único, si puede ser nulo, etc.

3. Crear y gestionar la base de datos

Crear la base de datos y las tablas

Después de definir los modelos, puedes utilizar los métodos proporcionados por SQLAlchemy para crear la base de datos y las tablas.

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

db.create_all(): crea las tablas correspondientes a todos los modelos definidos en el contexto actual.

4. Operaciones CRUD básicas

Crear registros

Ejemplo

@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 'User added!'

db.session.add(new_user): agregar el nuevo objeto de usuario a la sesión.

db.session.commit(): confirmar la transacción y guardar los cambios en la base de datos.

Leer registros

Ejemplo

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

User.query.all(): consultar todos los registros de usuarios.

Actualizar registros

Ejemplo

@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 'User updated!'
    return 'User not found!'

User.query.get(user_id): consultar un único registro de usuario mediante su clave primaria.

Actualizar los valores de los campos y confirmar la transacción.

Eliminar registros

Ejemplo

@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 'User deleted!'
    return 'User not found!'

db.session.delete(user): eliminar el registro de usuario y confirmar la transacción.

5. Operaciones de consulta

SQLAlchemy ofrece potentes funciones de consulta; puedes realizar varias operaciones de consulta mediante el objeto de consulta.

Consultas básicas

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

filter_by(): filtrar registros según el valor de un campo.

Consultas complejas

from sqlalchemy import or_

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

or_(): se utiliza para ejecutar condiciones de consulta complejas.

Ordenación y paginación

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

order_by(): ordenar por un campo especificado.

paginate(): consulta paginada.

6. Usar Flask-Migrate para migraciones

Flask-Migrate es una extensión para migraciones de bases de datos, basada en Alembic, que te ayuda a gestionar el control de versiones de la base de datos.

Instalar Flask-Migrate

pip install flask-migrate

Configurar Flask-Migrate

Código del archivo app.py:

from flask_migrate import Migrate

migrate = Migrate(app, db)

Inicializar la migración

flask db init

Crear el script de migración

flask db migrate -m "Initial migration."

Aplicar la migración

flask db upgrade
  • flask db init: inicializar el entorno de migración.
  • flask db migrate -m "message": crear el script de migración.
  • flask db upgrade: aplicar la migración a la base de datos.

7. Ejecutar SQL puro

Aunque SQLAlchemy ofrece funciones ORM, también puedes ejecutar sentencias SQL puras.

Ejemplo

@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(): ejecutar consultas SQL puras.

Conectar y operar con bases de datos MySQL en Flask

Conectar y operar con bases de datos MySQL en Flask generalmente implica usar SQLAlchemy o usar directamente el controlador de Python para MySQL. A continuación se detallan los pasos, incluyendo operaciones con Flask-SQLAlchemy y con el controlador de Python para MySQL directamente.

1. Conectar a MySQL usando Flask-SQLAlchemy

Flask-SQLAlchemy es una extensión de Flask que simplifica la configuración y el uso de SQLAlchemy. Para conectarte a MySQL, necesitas instalar Flask-SQLAlchemy y el controlador de MySQL.

Instalar las bibliotecas necesarias:

pip install flask-sqlalchemy mysqlclient
  • flask-sqlalchemy: la extensión de SQLAlchemy para Flask.
  • mysqlclient: el controlador de Python para la base de datos MySQL.

Configurar Flask-SQLAlchemy

Código del archivo app.py:

Ejemplo

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: establecer la URI de conexión a la base de datos, con el formatomysql://username:password@localhost/dbname。

  • username: nombre de usuario de MySQL.
  • password: contraseña de MySQL.
  • localhost: dirección del host de MySQL (normalmente local comolocalhost）。
  • dbname: nombre de la base de datos.

Definir modelos y realizar operaciones básicas

Código del archivo app.py:

Ejemplo

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()  # Crear las tablas de la base de datos
    app.run(debug=True)

Usar directamente el controlador de Python para MySQL

Si decides no usar SQLAlchemy y usar directamente el controlador de Python para MySQL, puedes utilizar las bibliotecas mysql-connector-python o PyMySQL.

Instalar mysql-connector-python:

pip install mysql-connector-python

Instalar PyMySQL (si eliges usar PyMySQL):

pip install PyMySQL

Conectar a MySQL usando mysql-connector-python

Código del archivo app.py:

Ejemplo

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 'User added!'

@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)

Conectar a MySQL usando PyMySQL, código del archivo app.py:

Ejemplo

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 'User added!'

@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)
Otras extensiones