CRUD en Flask

Configuración inicial

Crear el entorno virtual
python -m venv .venv

Iniciar entorno virtual
.venv/Scripts/actívate

Instalar las dependencias necesarias
pip install flask flask_cors flask_sqlalchemy pymysql jsonify

**Generar archivo que contiene una lista de todas las dependencias y sus versiones instaladas en tu entorno virtual de Python
pip freeze > requirements.txt

Creación del aplicación inical

Crear un archivo python con el nombre de app.py, puedes colocar el nombre que desees.

from flask import Flask
from flask_cors import CORS
from flask_sqlalchemy import SQLAlchemy

app = Flask(__name__)
# -------------------------------------------------------------------- #
# COnfiguración CORS
# -------------------------------------------------------------------- #
CORS(app)

# -------------------------------------------------------------------- #
# Configuración SQLAlchemy
# -------------------------------------------------------------------- #
app.config["SQLALCHEMY_DATABASE_URI"] = "mysql+pymysql://root:1066@localhost/test?charset=utf8mb4"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
app.config["SQLALCHEMY_ECHO"] = True 

# -------------------------------------------------------------------- #
# Inicialización de SQLAlchemy con la aplicación
# -------------------------------------------------------------------- #
db = SQLAlchemy(app)

# -------------------------------------------------------------------- #
# Importaciónd de los blueprints de cada una de las rutas
# -------------------------------------------------------------------- #
from routes import categorias, productos

# -------------------------------------------------------------------- #
# Creación de las tablas en la base de datos a través de los modelos con SQLAlchemy
# -------------------------------------------------------------------- #
with app.app_context():
    from models import *
    db.create_all()

# -------------------------------------------------------------------- #
# Registro de Blueprints rutas publicas
# -------------------------------------------------------------------- #
app.register_blueprint(productos)
app.register_blueprint(categorias)

@app.route("/")
def hello_world():
    return "<p>Hola, Soy una pequeña api</p>"

Comando para correr la aplicación inical
flask --app app run

Creación de los modelos

from sqlalchemy import Float, ForeignKey, Integer, Numeric, String, Boolean, DateTime, func
from sqlalchemy.orm import relationship, Mapped, mapped_column
from app import db  # Importa db desde app.py

# -------------------------------------------------------------------------------------------------------- #
# Modelos para las categorias
# -------------------------------------------------------------------------------------------------------- #

class Categorias(db.Model):  # Usa db.Model como base para las clases
    __tablename__ = 'categorias'

    id_categorias: Mapped[int] = mapped_column(Integer, primary_key=True)
    nombre: Mapped[str] = mapped_column(String(255), nullable=False)
    url_imagen: Mapped[str] = mapped_column(String(255))
    is_activo: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    fecha_creacion: Mapped[DateTime] = mapped_column(DateTime, default=func.now())
    fecha_actualizacion: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())

    productos: Mapped[list["Productos"]] = relationship(back_populates="categorias", cascade="all, delete-orphan")  # type: ignore

    def to_dict(self):
        return {
            "nombre": self.nombre,
            "url_imagen": self.url_imagen,
            "is_activo": self.is_activo,
            "fecha_creacion": self.fecha_creacion,
            "fecha_actualizacion": self.fecha_actualizacion
        }

    def __repr__(self):
        return f'<Nombre {self.nombre!r}, <Descripcion {self.descripcion!r}, <URLImagen {self.url_imagen!r}, <IsActivo {self.is_activo!r}>'


# --------------------------------------------------------------------- #
# Modelos para los productos
# --------------------------------------------------------------------- #

class Productos(db.Model):
    __tablename__ = "productos"

    id: Mapped[int] = mapped_column(primary_key=True)
    sku: Mapped[str] = mapped_column(String(255), unique=True, nullable=False)
    nombre: Mapped[str] = mapped_column(String(255), nullable=False)
    descripcion: Mapped[str] = mapped_column(String(1000))
    url_imagen: Mapped[str] = mapped_column(String(255))
    url_ficha_tecnica: Mapped[str] = mapped_column(String(255))
    unidad_producto: Mapped[str] = mapped_column(String(255), nullable=False)
    cantidad: Mapped[int] = mapped_column(Integer)
    precio: Mapped[float] = mapped_column(Float, nullable=False)
    is_promocion: Mapped[bool] = mapped_column(Boolean, nullable=False)
    stock: Mapped[int] = mapped_column(Integer, nullable=False)
    is_activo: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
    descuento: Mapped[float] = mapped_column(Numeric(10, 2))
    fecha_creacion: Mapped[DateTime] = mapped_column(DateTime, default=func.now())
    fecha_actualizacion: Mapped[DateTime] = mapped_column(DateTime, default=func.now(), onupdate=func.now())
    id_categorias: Mapped[int] = mapped_column(ForeignKey("categorias.id_categorias"))
    # ---------------------------------------------------------------- #
    # RELACIONES
    # Reladcion uno a muchos entre categorias y productos
    categorias: Mapped["Categorias"] = relationship(back_populates="productos")  # type: ignore
    # ---------------------------------------------------------------- #

    def set_initial_values(self):
        self.is_activo = True

    def to_dict(self):
        return {
            "id": self.id,
            "sku": self.sku,
            "nombre": self.nombre,
            "descripcion": self.descripcion,
            "url_imagen": self.url_imagen,
            "url_ficha_tecnica": self.url_ficha_tecnica,
            "unidad_producto": self.unidad_producto.value,
            "cantidad": self.cantidad,
            "precio": self.precio,
            "is_promocion": self.is_promocion,
            "stock": self.stock,
            "is_activo": self.is_activo,
            "descuento": self.descuento,
            "id_categorias": self.id_categorias,
            "fecha_creacion": self.fecha_creacion,
            "fecha_actualizacion": self.fecha_actualizacion
        }

    def __repr__(self):
        return f"<SKU {self.sku!r}>, <Nombre {self.nombre!r}, <URLImagen {self.url_imagen!r}, <URLFichaTecnica {self.url_ficha_tecnica!r}, <UnidadProducto {self.unidad_producto!r}, <Cantidad {self.cantidad!r}, <Precio {self.precio!r},<IsPromocion {self.is_promocion!r},<IsStock {self.stock!r},<IsActivo {self.is_activo!r}, <Descuento {self.descuento!r}, <FechaInicioDescuento {self.fecha_inicio_descuento!r}, <FechaFinDescuento {self.fecha_fin_descuento!r}>"

Consultar la documentación oficial de SQLALchemy 2.0 para los siguientes apartados:

Para las relaciones: Basic Relationship Patterns
Para el estilo declarativo de mappeado: Table Configuration with Declarative
Para los tipos de datos de base de datos: The Type Hierarchy

Creacion de las rutas

from flask import Blueprint, jsonify, request
from models import Categorias, Productos
from app import db

# --------------------------------------------------------------------- #
# Rutas para las categorias
# --------------------------------------------------------------------- #

categorias = Blueprint("categorias", __name__)

@categorias.get("/categorias")
def obtener_categorias():
    categorias = Categorias.query.all()
    lista_categorias = [
        {
            "id_categorias": categoria.id_categorias,
            "nombre": categoria.nombre,
            "url_imagen": categoria.url_imagen,
        }
        for categoria in categorias
    ]
    return jsonify(lista_categorias)


@categorias.get("/categorias/<int:id>")
def obtener_categoria_por_id(id):
    categoria = Categorias.query.get_or_404(id, description="Categoria no encontrada")
    if not categoria:
        return jsonify({"message": "Categoria no encontrada"}), 404
    return jsonify(
        {
            "id_categorias": categoria.id_categorias,
            "nombre": categoria.nombre,
            "url_imagen": categoria.url_imagen,
        }
    )

@categorias.post("/categorias")
def guardar_categroias():
    data = request.json
    nuevo_producto = Categorias( nombre=data['nombre'], 
                                url_imagen=data['url_imagen'], )
    db.session.add(nuevo_producto)
    db.session.commit()
    return jsonify({'message': 'Nueva categoria creada correctamente'}), 201

@categorias.patch('/categorias/<int:id>')
def actualizarcategroias(id):
    producto = Categorias.query.get(id)
    if not producto:
        return jsonify({'message': 'Producto no encontrada'}), 404
    data = request.json
    producto.nombre = data['nombre']
    producto.url_imagen = data['url_imagen']
    db.session.commit()
    return jsonify({'message': 'Categoria actualizada satisfactoriamente'}), 200

@categorias.delete('/categorias/<int:id>')
def eliminarcategroias(id):
    producto = Categorias.query.get(id)
    if not producto:
        return jsonify({'message': 'Producto no encontrada'}), 404
    db.session.delete(producto)
    db.session.commit()
    return jsonify({'message': 'La categoria ha sido eliminada satisactoriamnete'}), 200

# --------------------------------------------------------------------- #
# Rutas para los productos
# --------------------------------------------------------------------- #

productos = Blueprint("productos", __name__)

@productos.get("/productos")
def obtener_productos():
    productos = Productos.query.all()
    lista_productos = [
        {
            "id": producto.id,
            "sku": producto.sku,
            "nombre": producto.nombre,
            "descripcion": producto.descripcion,
            "url_imagen": producto.url_imagen,
            "url_ficha_tecnica": producto.url_ficha_tecnica,
            "unidad_producto": producto.unidad_producto.value,
            "cantidad": producto.cantidad,
            "precio": producto.precio,
            "is_promocion": producto.is_promocion,
            "stock": producto.stock,
            "descuento": producto.descuento,
            "is_activo": producto.is_activo,
            "id_categorias": producto.id_categorias,
            "fecha_inicio_descuento": producto.fecha_inicio_descuento,
            "fecha_fin_descuento": producto.fecha_fin_descuento,
        }
        for producto in productos
    ]
    return jsonify(lista_productos)


@productos.get("/productos/<int:id>")
def obtener_producto_por_id(id):
    producto = Productos.query.get_or_404(id, description="Producto no encontrado")
    if not producto:
        return jsonify({"message": "Producto no encontrada"}), 404
    return jsonify(
        {
            "id": producto.id,
            "nombre": producto.nombre,
            "descripcion": producto.descripcion,
            "url_imagen": producto.url_imagen,
            "url_ficha_tecnica": producto.url_ficha_tecnica,
            "unidad_producto": producto.unidad_producto.value,
            "cantidad": producto.cantidad,
            "precio": producto.precio,
            "is_promocion": producto.is_promocion,
            "stock": producto.stock,
            "descuento": producto.descuento,
            "is_activo": producto.is_activo,
            "id_categorias": producto.id_categorias,
            "fecha_inicio_descuento": producto.fecha_inicio_descuento,
            "fecha_fin_descuento": producto.fecha_fin_descuento,
        }
    )

@productos.post("/productos")
def guardar_productos():
    data = request.json
    nuevo_producto = Productos(sku=data['sku'], 
                                nombre=data['nombre'], 
                                descripcion=data['descripcion'], 
                                url_imagen=data['url_imagen'],
                                url_ficha_tecnica=data['url_ficha_tecnica'], 
                                unidad_producto=data['unidad_producto'],
                                cantidad=data['cantidad'], 
                                precio=data['precio'], 
                                is_promocion=data['is_promocion'], 
                                stock=data['stock'], 
                                descuento=data['descuento'],
                                id_categorias=data['id_categorias'])
    db.session.add(nuevo_producto)
    db.session.commit()
    return jsonify({'message': 'Nuevo producto creada correctamente'}), 201

@productos.patch('/productos/<int:id>')
def actualizar_producto(id):
    producto = Productos.query.get(id)
    if not producto:
        return jsonify({'message': 'Producto no encontrada'}), 404
    data = request.json
    producto.nombre = data['nombre']
    producto.descripcion = data['descripcion']
    producto.url_imagen = data['url_imagen']
    producto.url_ficha_tecnica = data['url_ficha_tecnica']
    producto.unidad_producto = data['unidad_producto']
    producto.cantidad = data['cantidad']
    producto.precio = data['precio']
    producto.is_promocion = data['is_promocion']
    producto.stock = data['stock']
    producto.descuento = data['descuento']
    producto.is_activo = data['is_activo']
    producto.id_categorias = data['id_categorias']
    db.session.commit()
    return jsonify({'message': 'Producto actualizado satisfactoriamente'}), 200

@productos.delete('/productos/<int:id>')
def eliminar_producto(id):
    producto = Productos.query.get(id)
    if not producto:
        return jsonify({'message': 'Producto no encontrada'}), 404
    db.session.delete(producto)
    db.session.commit()
    return jsonify({'message': 'El producto ha sido eliminado satisfactoriamnete'}), 200

Blog

G00Z

Configuración inicial

Creación del aplicación inical

Creación de los modelos

Creacion de las rutas

Join Our Newsletter. No Spam, Only the good stuff.

Related