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>"
Enter fullscreen mode
Exit fullscreen mode

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}>"
Enter fullscreen mode
Exit fullscreen mode

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

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
Enter fullscreen mode
Exit fullscreen mode

Discover more from Coursity

Subscribe to get the latest posts sent to your email.

Leave a Comment

Your email address will not be published. Required fields are marked *

Discover more from Coursity

Subscribe now to keep reading and get access to the full archive.

Continue reading

Scroll to Top
From windows vps servers to linux vps servers and gaming servers, our portfolio has something for everyone. Gets answers to the most frequently asked questions about opening a new restaurant, managing a.