Despliegue de MySQL en Azure

Para poder almacenar datos en almacenamiento persistente existe una gran variedad de bases de datos relacionales como MySQL, PostgreSQL, SQL Server, Oracle, SQLite, etc. 

Podemos instalar MySQL Community Server en nuestro ordenador, o simplemente desplegar un MySQL en Azure para poder conectarnos desde Python y empezar a almacenar resultados.

En Azure, buscamos el servicio: Servidores flexibles de Azure Database for MySQL. 

Una vez encontrado el servicio y pulsada la opción de creación, nos deja configurar los datos de autenticación y la configuración de la base de datos.

Una vez completados los campos pulsamos en Revisar y crear. Tras el despliegue ya nos proporciona los datos de acceso, como el nombre del servidor, el usuario y la contraseña que hayamos puesto:

Ahora que ya tenemos la base de datos desplegada, para poder conectarnos desde Python necesitamos permitir el acceso desde nuestra IP pública y descargar el certificado TLS para usarlo en la conexión desde Python:

Para conectar desde python utilizaremos la biblioteca mysql connector, y probaremos a ejecutar una sentencia SQL básica para ver qué bases de datos hay instaladas en el servidor mysql. Primero instalamos el paquete mysql connector:

pip install mysql-connector-python

Después creamos un archivo python o jupyter notebook, desde donde podemos probar a conectarnos a la base de datos, utilizando el certificado pem de Microsoft:

import mysql.connector as con

database = con.connect(
    host="mysql-openai-testing.mysql.database.azure.com",
    port=3306,
    user="dev_user",
    password="tu_password",
    # este archivo debe estar en el mismo directorio que este script
    ssl_ca="DigiCertGlobalRootG2.crt.pem"  
)

cursor = database.cursor()

sql = """
SHOW DATABASES;
"""
cursor.execute(sql)

for name in cursor:
    print(name)

cursor.close()
database.close()

Como resultado encontraremos unas bases de datos ya predefinidas:

('information_schema',)
('mysql',)
('performance_schema',)
('sys',)

Si queremos crear una nueva base de datos podemos ejecutar el siguiente código SQL en Python:

cursor = database.cursor()
sql = """
DROP DATABASE IF EXISTS python_mysql;
"""
cursor.execute(sql)

sql = """
CREATE SCHEMA python_mysql 
DEFAULT CHARACTER SET utf8 ;
"""
cursor.execute(sql)

Como alternativa visual, podemos acceder desde el propio panel de Azure o podemos conectarnos al servidor utilizando el programa MySQL Workbench. Ejemplo desde Azure:

Creación de modelos con SQLAlchemy en MySQL

En esta sección, aprenderemos a definir modelos de persistencia utilizando SQLAlchemy para interactuar con nuestra base de datos MySQL en Azure. Esto nos permitirá manipular los datos de forma más intuitiva y orientada a objetos dentro de nuestras aplicaciones Python, en lugar de tener que crear y procesar manualmente las sentencias SQL.

SQLAlchemy es una biblioteca (o librería) de Python que facilita la interacción con bases de datos relacionales. Combina dos componentes principales:

• SQLAlchemy Core:
  • Proporciona una abstracción de menor nivel sobre SQL, permitiendo construir consultas de manera más estructurada y portable que escribir directamente SQL “a mano”.
  • Utiliza un enfoque llamado “Expression Language” para definir estructuras SQL (tablas, columnas, condiciones, etc.) a través de objetos de Python.
  • Permite a los desarrolladores crear y ejecutar consultas SQL de una forma más organizada, sin sacrificar el control detallado sobre el SQL resultante.
• SQLAlchemy ORM (Object-Relational Mapper):
  • Construido sobre el “core”, provee herramientas para mapear clases de Python a tablas de la base de datos (y sus atributos a columnas).
  • Permite operar con las filas de la base de datos como si fueran objetos de Python (instancias de clases), ocultando buena parte de la complejidad del manejo de SQL crudo.
  • Facilita la definición de relaciones (por ejemplo, uno a muchos, muchos a muchos, etc.) mediante propiedades y anotaciones en el modelo.
  • Integra un sistema de sesiones (Session) para agrupar operaciones y controlar transacciones, lo que simplifica y asegura la consistencia de los datos durante las operaciones de CRUD (Crear, Leer, Actualizar, Borrar).

SQLAlchemy, en el ecosistema Python, se integra bien con múltiples frameworks (Flask, FastAPI, etc.). En el caso del framework Django ya incorpora su propio ORM conocido como Django ORM.

Para empezar, asegurémonos de tener instaladas las dependencias necesarias. Ejecutamos el siguiente comando para instalar SQLAlchemy y el controlador PyMySQL, que permite la comunicación entre SQLAlchemy y MySQL:

pip install sqlalchemy pymysql

A continuación, configuramos la cadena de conexión para conectarnos a nuestra base de datos MySQL utilizando SQLAlchemy:

from sqlalchemy import create_engine

usuario = 'dev_user'
contraseña = "tu_password"
servidor = 'mysql-openai-testing.mysql.database.azure.com'
base_datos = 'python_mysql'
ruta_certificado = 'DigiCertGlobalRootG2.crt.pem'

cadena_conexion = (
    f"mysql+pymysql://{usuario}:{contraseña}@{servidor}/{base_datos}"
    f"?ssl_ca={ruta_certificado}&ssl_verify_cert=true"
)

engine = create_engine(cadena_conexion, echo=True)

En este fragmento, hemos creado un engine que maneja la comunicación con la base de datos. El parámetro echo=True es opcional y permite visualizar las consultas SQL generadas por SQLAlchemy, lo cual es útil para depuración.

Ahora, definimos una clase base a partir de la cual construiremos nuestros modelos. Utilizamos declarative_base para este propósito:

from sqlalchemy.orm import declarative_base

Base = declarative_base()

Con la clase base establecida, podemos definir nuestros modelos que representan tablas en la base de datos. Por ejemplo, creamos un modelo para una tabla de clientes:

from sqlalchemy import Column, Integer, String, DateTime
import datetime

class Cliente(Base):
    __tablename__ = 'clientes'

    id = Column(Integer, primary_key=True)
    nombre = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    fecha_registro = Column(DateTime, default=datetime.datetime.now)

    def __repr__(self):
        return f"<Cliente(nombre='{self.nombre}', email='{self.email}')>"

En este modelo, cada atributo corresponde a una columna en la tabla clientes. Hemos definido tipos de datos como String e Integer, y establecemos restricciones como nullable=False y unique=True para garantizar la integridad de los datos.

Para crear las tablas en la base de datos basadas en nuestros modelos, utilizamos el método create_all del metadata:

# Creación de tablas en la base de datos
Base.metadata.create_all(engine)

Este comando genera las tablas que aún no existen en la base de datos según los modelos definidos.

Para comprobarlo, podemos entrar en MySQL Workbench y conectarnos a la base de datos, donde podemos realizar una sentencia select y verificar la estructura de la tabla:

Para interactuar con la base de datos, necesitamos una sesión que nos permita realizar operaciones CRUD (Crear, Leer, Actualizar, Eliminar). Configuramos la sesión de la siguiente manera:

from sqlalchemy.orm import sessionmaker

Session = sessionmaker(bind=engine)
session = Session()

Con la sesión iniciada, podemos agregar un nuevo cliente a la base de datos:

# Creación de un nuevo cliente
nuevo_cliente = Cliente(nombre='María López', email='maria.lopez@example.com')

# Agregar el cliente a la sesión
session.add(nuevo_cliente)

# Confirmar la transacción
session.commit()

Es importante confirmar la transacción con commit() para que los cambios se reflejen en la base de datos. Si deseamos consultar los clientes existentes, podemos realizar una consulta:

# Consulta de todos los clientes
lista_clientes = session.query(Cliente).all()
for cliente in lista_clientes:
    print(cliente)

Para filtrar resultados, utilizamos métodos como filter_by o filter:

# Buscar un cliente por nombre
cliente = session.query(Cliente).filter_by(nombre='María López').first()
print(cliente)

Si necesitamos actualizar la información de un cliente, modificamos los atributos y confirmamos la transacción:

# Actualizar el email de un cliente
cliente.email = 'maria.lopez2025@example.com'
session.commit()

Para eliminar un cliente de la base de datos:

# Eliminar un cliente
session.delete(cliente)
session.commit()

Al finalizar las operaciones, es recomendable cerrar la sesión para liberar los recursos:

# Cerrar la sesión
session.close()

Con estos pasos, hemos establecido una estructura de modelos que nos permite interactuar de manera eficiente con nuestra base de datos MySQL utilizando SQLAlchemy. Esto facilita el desarrollo de aplicaciones más robustas y mantenibles al trabajar con datos de forma orientada a objetos.

Ahora que sabemos cómo almacenar objetos creados a partir de clases Python, podemos hacer lo mismo con los objetos creados con las clases obtenidas de salidas estructuradas de APIs de IA.

API REST con Pydantic en Flask y SQLAlchemy

En esta sección, integraremos SQLAlchemy y Pydantic en una aplicación Flask para construir una API REST que permita almacenar información en la base de datos MySQL desplegada en Azure.

Esta integración nos permitirá gestionar datos de manera eficiente dentro de una aplicación web con Python. Primero, definiremos las tecnologías que vamos a usar.

Flask es un microframework para Python que permite construir aplicaciones web y APIs de forma muy simple. Se denomina “microframework” porque proporciona solo los componentes esenciales (manejo de peticiones, rutas, etc.) y no obliga a utilizar librerías de validación ni de acceso a datos específicas.

Flask es extremadamente flexible: puedes escoger tus propios componentes para la parte de persistencia (SQLAlchemy, Mongo, etc.) o validación (Pydantic, Marshmallow, etc.).

Por otro lado, Pydantic es una librería de validación de datos basada en las anotaciones de tipo de Python, que nos sirve para gestionar la serialización de objetos a JSON y viceversa. Nos sirve tanto para gestionar JSON en la API REST de Flask como para gestionar JSON en las salidas estructuradas de las APIs de IA que se quieran integrar en la aplicación Flask.

Ahora, instalamos las dependencias necesarias ejecutando:

pip install Flask Flask_SQLAlchemy pydantic

El objetivo es crear una aplicación Flask capaz de recibir clientes vía API REST y almacenarlos en la base de datos MySQL permitiendo realizar las operaciones CRUD sobre ellos.

Crear app.py y configuración de base de datos

A continuación, creamos el archivo principal de nuestra aplicación, app.py, y escribimos:

from flask import Flask, request, jsonify
from pydantic import BaseModel, ValidationError, ConfigDict
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.exc import IntegrityError
import datetime

app = Flask(__name__)

En el mismo archivo, seguimos agregando nuevo código, configuramos la aplicación para conectarse a la base de datos MySQL en Azure. 

Añadimos la cadena de conexión en la configuración de Flask:

usuario = 'dev_user'
contraseña = "tu_password"
servidor = 'mysql-openai-testing.mysql.database.azure.com'
base_datos = 'python_mysql'
ruta_certificado = 'DigiCertGlobalRootG2.crt.pem'

cadena_conexion = (
    f"mysql+pymysql://{usuario}:{contraseña}@{servidor}/{base_datos}"
    f"?ssl_ca={ruta_certificado}&ssl_verify_cert=true"
)

app.config['SQLALCHEMY_DATABASE_URI'] = cadena_conexion
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

En este fragmento, establecemos la conexión segura mediante SSL y desactivamos el seguimiento de modificaciones con SQLALCHEMY_TRACK_MODIFICATIONS para mejorar el rendimiento.

Crear schema Pydantic y modelo de SQL Alchemy

Para poder enviar y recibir datos en JSON de forma cómoda usamos Pydantic, que nos permite crear objetos Python a partir de JSON. Creamos el esquema para cliente:

class ClienteSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int | None
    nombre: str
    email: str
    fecha_registro: datetime.datetime | None

Definimos nuestros modelos extendiendo db.Model. Por ejemplo, el modelo Cliente:

class Cliente(db.Model):
    __tablename__ = 'clientes'

    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(50), nullable=False)
    email = db.Column(db.String(100), unique=True, nullable=False)
    fecha_registro = db.Column(db.DateTime, default=db.func.current_timestamp())

    @classmethod
    def from_pydantic(cls, pydantic_model: ClienteSchema):
        """Crea una instancia de SQLAlchemy a partir de un objeto Pydantic."""
        return cls(
            id=pydantic_model.id,
            nombre=pydantic_model.nombre,
            email=pydantic_model.email,
            fecha_registro=pydantic_model.fecha_registro
        )

Después continuamos creando los métodos para la API REST. Los métodos permitirán:

• Ver todos los clientes en base de datos
• Ver un cliente por su id
• Crear un nuevo cliente
• Actualizar un cliente existente
• Borrar un cliente por su id

Métodos GET para obtener clientes:

Para obtener todos los clientes con una solicitud GET:

@app.route('/clientes', methods=['GET'])
def obtener_clientes():

    clientes = Cliente.query.all()
    # Convertir cada objeto SQLAlchemy a dict validado por Pydantic
    clientes_validados = [
        ClienteSchema.model_validate(cliente).model_dump()
        for cliente in clientes
    ]
    return jsonify(clientes_validados)

Para obtener un cliente específico por ID:

@app.route('/clientes/<int:cliente_id>/', methods=['GET'])
def obtener_cliente(cliente_id):

    cliente = Cliente.query.get_or_404(cliente_id)
    return ClienteSchema.model_validate(cliente).model_dump()

Utilizamos get_or_404 para manejar automáticamente el caso en que el cliente no exista, devolviendo un error 404.

Método POST para crear clientes

Ahora, definimos las rutas para nuestra API. Para crear un nuevo cliente mediante una solicitud POST:

@app.route('/clientes', methods=['POST'])
def crear_cliente():

    try:
        data = request.get_json()
        cliente_data = ClienteSchema.model_validate(data)  # Validar con Pydantic
        nuevo_cliente = Cliente.from_pydantic(cliente_data)
        db.session.add(nuevo_cliente)
        db.session.commit()
        return ClienteSchema.model_validate(nuevo_cliente).model_dump(), 201

    except ValidationError as e:
        return jsonify({"error": "Validación fallida", "detalles": e.errors()}), 422
    except IntegrityError as e:
        db.session.rollback()
        return jsonify({"error": "Error de integridad (posible email duplicado)", "detalles": str(e)}), 400
    except Exception as e:
        return jsonify({"error": "Error interno", "detalles": str(e)}), 500

En este código, recibimos datos en formato JSON, creamos un objeto Cliente y lo guardamos en la base de datos. 

Es crucial confirmar la transacción con db.session.commit() para persistir los cambios.

También podemos gestionar los errores con try except proporcionando lógica específica para cada tipo de error mejorando las respuestas de la API REST.

Método PUT para actualizar cliente

Para actualizar un cliente con una solicitud PUT:

@app.route('/clientes/<int:cliente_id>/', methods=['PUT'])
def actualizar_cliente(cliente_id):

    try:
        data = request.get_json()
        cliente_data = ClienteSchema.model_validate(data)  # Validar con Pydantic
        cliente_db = Cliente.query.get_or_404(cliente_id)
        cliente_db.nombre = cliente_data.nombre
        cliente_db.email = cliente_data.email
        db.session.commit()
        return ClienteSchema.model_validate(cliente_db).model_dump(), 200

    except ValidationError as e:
        return jsonify({"error": "Validación fallida", "detalles": e.errors()}), 422
    except IntegrityError as e:
        db.session.rollback()
        return jsonify({"error": "Error de integridad (posible email duplicado)", "detalles": str(e)}), 400
    except Exception as e:
        return jsonify({"error": "Error interno", "detalles": str(e)}), 500

En este fragmento, actualizamos los campos nombre y email del cliente, con gestión de errores.

Método DELETE para borrar cliente

Para eliminar un cliente con una solicitud DELETE:

@app.route('/clientes/<int:cliente_id>/', methods=['DELETE'])
def eliminar_cliente(cliente_id):

    try:
        cliente_db = Cliente.query.get_or_404(cliente_id)
        db.session.delete(cliente_db)
        db.session.commit()
        return jsonify({"message": "Cliente eliminado"}), 200
    except Exception as e:
        return jsonify({"error": "Error interno", "detalles": str(e)}), 500

Finalmente, escribimos el código para ejecutar la aplicación, invocando el método create_all para generar las tablas en caso de que no existan:

if __name__ == '__main__':
    # Crear la tabla 'clientes' en caso de que no exista
    with app.app_context():
        db.create_all()

    app.run(debug=True)

Activamos el modo debug para facilitar la detección y solución de posibles errores durante el desarrollo.

Con estos pasos, hemos integrado SQLAlchemy, MySQL y Flask con Pydantic para construir una API que permite operaciones CRUD sobre nuestros datos. Esta estructura es fundamental para desarrollar aplicaciones web robustas y escalables que interactúen con bases de datos en la nube.

Con las tecnologías implementadas es posible interactuar con la API de OpenAI o cualquier IA que soporte salidas estructuradas y almacenarlas en base de datos utilizando SQL Alchemy.

Una alternativa al microframework Flask son los frameworks FastAPI y Django.

FastAPI es un framework web para Python diseñado para crear APIs de manera rápida y eficiente lanzado en 2018. Está basado en  Starlette y Pydantic, utiliza tipado estático, genera documentación de API automática y tiene soporte para asincronía, se caracteriza por tener alto rendimiento.

Django es un framework web de alto nivel para Python lanzado en 2005 y que fomenta el desarrollo rápido y el diseño limpio. Viene con una amplia gama de funcionalidades incluidas a diferencia de Flask.

Por otro lado, Flask fue lanzado en 2010 y es conocido por su simplicidad y flexibilidad, tiene un ecosistema de bibliotecas muy amplio que lo hace ideal para microservicios y aplicaciones pequeñas y medianas.

Fuera del ecosistema Python también existen otros frameworks de backend como Spring Boot para Java, ASP.NET para C#, Laravel para PHP, NestJS para TypeScript, cada uno con ORMs disponibles para conectarse a bases de datos de forma similar a la que hemos explicado. La elección de un framework u otro dependerá de los requisitos del proyecto a desarrollar.

Desarrollo de aplicación API REST para generar publicaciones de LinkedIn con openai

En esta sección vamos a desarrollar una API REST en Flask que sea capaz de generar posts para LinkedIn usando la API de OpenAI.

Como resultado, podremos enviar una petición a Flask con un prompt que la aplicación usará para pedir a OpenAI que genere publicaciones de LinkedIn utilizando salidas estructuradas que almacenaremos en base de datos con SQL Alchemy. De esta forma tendremos una tabla en base de datos con las publicaciones de LinkedIn con campos como título, contenido, url de recurso externo, prompt para generación de imagen, fecha de creación, hashtags.

Creación de configuración

Creamos una carpeta flask por ejemplo, en la que crearemos un archivo config.py que tendrá la configuración para conectarse a la base de datos MySQL:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy

usuario = 'dev_user'
contraseña = "tu_password"
servidor = 'mysql-openai-testing.mysql.database.azure.com'
base_datos = 'python_mysql'
ruta_certificado = 'DigiCertGlobalRootG2.crt.pem'

cadena_conexion = (
    f"mysql+pymysql://{usuario}:{contraseña}@{servidor}/{base_datos}"
    f"?ssl_ca={ruta_certificado}&ssl_verify_cert=true"
)

def create_app():
    app = Flask(__name__)
    app.config['SQLALCHEMY_DATABASE_URI'] = cadena_conexion
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    return app

db = SQLAlchemy()

En este archivo configuramos la conexión a base de datos e inicializamos SQLAlchemy. 

Por simplicidad hemos mantenido las credenciales en el archivo, pero por seguridad se puede extraer a un archivo .env y cargarlas con la biblioteca dotenv para python.

Además, en este archivo podemos agregar otras configuraciones que necesitemos.

Creación de esquemas Pydantic

Para interactuar con la API de Open AI con salidas estructuradas, usaremos esquemas de Pydantic. 

Para ello, creamos el archivo schemas.py con el siguiente código:

from pydantic import BaseModel, ConfigDict
from typing import List


class LinkedInPostSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    title: str
    content: str
    hashtags: str
    link: str
    image_prompt: str
    
class LinkedInPostSchemas(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    linkedin_posts: List[LinkedInPostSchema]

En este archivo hemos creado dos esquemas, LinkedInPostSchema representa una publicación de linkedin que será generada por un modelo de IA. Por otro lado, LinkedInPostSchemas es un esquema para envolver una lista de objetos LinkedInPostSchema así podemos trabajar con más de una publicación al mismo tiempo.

Se puede agregar nuevos campos con información que queremos que genere la IA, en este ejemplo, para mantenerlo simple estamos solicitado un título, el contenido de la publicación, un string con una lista de hashtags, un enlace a un recurso externo y un prompt para generar una imagen por si a futuro queremos también generar una imagen por IA para dicha publicación.

Creación de modelos SQLAlchemy

Ahora necesitamos el modelo de SQLAlchemy que generará la tabla de base de datos, y donde almacenaremos los resultados obtenidos.

Creamos un archivo models.py con el siguiente contenido:

from config import db

class LinkedInPost(db.Model):
    __tablename__ = "linkedin_posts"

    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(255), nullable=False)
    content = db.Column(db.Text, nullable=False)
    hashtags = db.Column(db.String(500), nullable=True)
    link = db.Column(db.String(500), nullable=True)
    image_prompt = db.Column(db.String(1000), nullable=True)
    image_url = db.Column(db.String(500), nullable=True)
    created_at = db.Column(db.DateTime, default=db.func.current_timestamp())

    def __repr__(self):
        return f"<LinkedInPost id={self.id} title={self.title}>"

    @classmethod
    def from_pydantic(cls, pydantic_model):
        return cls(
            title=pydantic_model.title,
            content=pydantic_model.content,
            image_prompt=pydantic_model.image_prompt,
            hashtags=pydantic_model.hashtags,
            link=pydantic_model.link,
            image_url=None
        )

Este archivo define todos los modelos de base de datos que vamos a usar, en este caso un modelo LinkedInPost para almacenar las publicaciones en base de datos. Además, con un método from_pydantic para facilitarnos la conversión de objeto esquema de Pydantic a objeto modelo de SQLAlchemy.

Desarrollar la app con Jinja

Ahora que ya tenemos todas las piezas, creamos el archivo app.py donde definiremos las rutas de API REST que permitirán ver todas las publicaciones existentes en base de datos y generar nuevas utilizando IA.

Primero añadiremos las importaciones necesarias e inicializaremos la app junto con el cliente OpenAI:

from flask import jsonify, request
from openai import OpenAI
from schemas import LinkedInPostSchema, LinkedInPostSchemas
from config import create_app, db
from models import LinkedInPost

app = create_app()
db.init_app(app)
client = OpenAI(api_key='tu_api_key')

Una vez tenemos la primera parte ya podemos empezar a crear los métodos de API REST para atender las solicitudes HTTP que interactúen con la IA. Empezamos creando un método GET que simplemente devuelva las publicaciones de LinkedIn que ya existan en base de datos, ordenadas por fecha de manera descendente, para ver primero las más recientes. Lo agregamos en el mismo archivo app.py a continuación del código que ya tenemos:

@app.route("/", methods=["GET"])
def ver_publicaciones():
    linkedin_posts = LinkedInPost.query.order_by(LinkedInPost.created_at.desc()).all()
    posts_schemas = [
        LinkedInPostSchema.model_validate(post).model_dump()
        for post in linkedin_posts
    ]
    return jsonify(posts_schemas), 200

Este método obtiene las publicaciones de base de datos, las convierte a objetos Pydantic y las devuelve. Si se desea se puede ignorar el uso de Pydantic en este método y gestionar las respuestas JSON de otra manera.

Podemos probar el método lanzando una petición HTTP por consola o con Postman, cuando tengamos datos mostrará una salida similar a esta:

Después creamos un método POST que será el que reciba un prompt del usuario y se encargue de solicitar a la IA que genere los contenidos para LinkedIn.

@app.route("/", methods=["POST"])
def crear_publicaciones():
    data = request.get_json()
    user_prompt = data['prompt'] if data else ""
    
    if not user_prompt:
        return jsonify({"message": "No se ha proporcionado un prompt"}), 400

    completion = client.beta.chat.completions.parse(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": 
                '''
                Eres un asistente de creación de contenidos para marketing en LinkedIn, cada publicación tendrá la siguiente estructura:
                    título (obligatorio),
                    contenido (obligatorio) (sin hashtags),
                    hashtags (obligatorio),
                    link (opcional) (url a un recurso público que permita profundizar más, como una documentación oficial por ejemplo),
                    image_prompt (opcional) (un prompt para generar una imagen con IA que se debería incluir en la publicación relacionada con el tema).
                '''},
            {"role": "user", "content": user_prompt},
        ],
        response_format=LinkedInPostSchemas,
    )
    response = completion.choices[0].message.parsed
    print(response)
    print(response.linkedin_posts)
    linkedin_posts = [LinkedInPost.from_pydantic(post) for post in response.linkedin_posts]
    db.session.add_all(linkedin_posts)
    db.session.commit()
    
    posts_schemas = [
        LinkedInPostSchema.model_validate(post).model_dump()
        for post in linkedin_posts
    ]
    return jsonify(posts_schemas), 201

En este ejemplo el método crear_publicaciones llama a la IA utilizando una salida estructurada que hemos definido en schemas.py, almacena los resultados en base de datos utilizando SQLAlchemy y los devuelve.

Podemos probar en Postman a enviar un prompt a este método:

Para poder ejecutar la aplicación y que se genere automáticamente las tablas de base de datos necesitamos agregar al final del archivo app.py el siguiente contenido:

if __name__ == '__main__':
    with app.app_context():
        db.create_all()

    app.run(debug=True)

Con este código ya tenemos un ejemplo de aplicación con IA integrada para generar contenido en base de datos de forma estructurada.

Si queremos, podemos extender este ejemplo para hacerlo más sofisticado y pedir a la IA que también genere una imagen utilizando el campo image_prompt y después descargamos y almacenamos la imagen en almacenamiento cloud como por ejemplo Azure Storage y la url de la imagen la guardamos en nuestra tabla junto al resto de datos que ya tenemos.