2.9. Autogenerar modelos SQLAlchemy

Nota

Propósito: Generador automático de código fuente de modelos SQLAlchemy.

sqlacodegen es una herramienta que lee la estructura de una base de datos existente y genera el código de modelos SQLAlchemy apropiado, usando el estilo declarativo si es posible.

Esta herramienta se escribió como reemplazo de sqlautocode, que sufría varios problemas (incluida, entre otras, la incompatibilidad con Python 3 y la última versión de SQLAlchemy).

2.9.1. Características

  • Soporta SQLAlchemy 0.8.x - 1.4.x.

  • Produce código declarativo que casi parece escrito a mano.

  • Produce código compatible con PEP 8.

  • Determina con precisión las relaciones, incluidos muchos a muchos, uno a uno.

  • Detecta automáticamente la herencia de tablas unidas.

  • Excelente cobertura de prueba.

2.9.2. Instalación

Para instalar la herramienta sqlacodegen debe seguir el siguientes paso, el cual a continuación se presentan el correspondiente comando de tu sistema operativo:

pip3 install sqlacodegen

Puede probar si la instalación se realizo correctamente, ejecutando el siguiente comando correspondiente a tu sistema operativo:

python3 -c "import sqlacodegen ; print(sqlacodegen.__name__)"

Si muestra el nombre del paquete como sqlacodegen, tiene correctamente instalada la herramienta. Con esto, ya tiene todo listo para continuar.

2.9.2.1. Estructura de archivos

Para crear la estructura de archivos del proyecto sqlacodegen debe ejecutar los siguientes comandos:

Crear el directorio ~/proyectos/sqlacodegen/sistema con el siguiente comando:

mkdir -p ~/proyectos/sqlacodegen/sistema && cd $_

El comando anterior crea la siguiente estructura de directorios:

proyectos/
└── sqlacodegen/
    └── sistema/

Si tiene la estructura de archivo previa, entonces puede continuar con la siguiente sección.

2.9.3. Conexión al Engine

Luego configura el engine (cadena de conexión) para la DB respectiva. Para esto se explican algunas configuraciones, para SQLite, MySQL y PostgreSQL a continuación:

Truco

Cada cadena de conexión necesita que se instale el módulo dependiente.

2.9.3.1. SQLite

Para configurar el engine con SQLite debe definir la cadena de conexión que esta compuesto por varios parámetros.

Los parámetros deben ser reemplazadas con sus propios datos en la línea de comando:

sqlacodegen --generator declarative sqlite:///{DB_PATH}/{DB_FILE} --outfile file.py

A continuación se presentan el correspondiente comando de tu sistema operativo:

sqlacodegen --generator declarative sqlite:///db.sqlite3 --outfile models.py

2.9.3.2. MySQL

Para configurar el engine con MySQL debe definir la cadena de conexión que esta compuesto por varios parámetros, los cuales deben ser reemplazadas con sus propios datos en la línea de comando con el comando sqlacodegen.

Truco

Para conectarte al servidor MySQL necesite el módulo PyMySQL.

Luego ya teniendo instalado el módulo PyMySQL debe ejecutar el siguiente comando de sqlacodegen, los parámetros deben ser reemplazadas con sus propios datos en la línea de comando:

sqlacodegen --generator declarative mysql+pymysql://USER:PASSW@HOST:PORT/DB --outfile file.py

A continuación se presentan el correspondiente comando de tu sistema operativo:

sqlacodegen --generator declarative mysql+pymysql://root:root@localhost:3306/sistema \
  --outfile models.py

2.9.3.3. PostgreSQL

Para configurar el engine con PostgreSQL debe definir la cadena de conexión que esta compuesto por varios parámetros, los cuales deben ser reemplazadas con sus propios datos en la línea de comando con el comando sqlacodegen.

Truco

Para conectarte al servidor PostgreSQL necesite el módulo psycopg2.

Luego ya teniendo instalado el módulo psycopg2 debe ejecutar el siguiente comando de sqlacodegen, los parámetros deben ser reemplazadas con sus propios datos en la línea de comando:

sqlacodegen --generator declarative postgresql://USER:PASSW@HOST:PORT/DB --outfile file.py

A continuación se presentan el correspondiente comando de tu sistema operativo:

sqlacodegen --generator declarative postgresql://root:root@localhost:3306/sistema \
  --outfile models.py

Si necesita más tipos de cadenas de conexión o engine, puede consultar el apartado Engine Configuration de la documentación oficial de SQLAlchemy.

2.9.4. Generación de modelos

Luego de haber definido la cadena de conexión a la base de datos, puede realizar la generación de modelos SQLAlchemy, ejecutando el siguiente comando, el cual a continuación se presentan el correspondiente comando de tu sistema operativo:

sqlacodegen --generator declarative sqlite:///sistema.db --outfile models.py

El anterior comando al ejecutar debe generar un módulo python llamado models.py que contiene el siguiente código:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
from sqlalchemy import Column, Date, Enum, ForeignKey, Integer, String
from sqlalchemy.orm import declarative_base, relationship

Base = declarative_base()


class Estados(Base):
    __tablename__ = "estados"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(25), nullable=False)
    codigo = Column(String(2), nullable=False)

    ciudades = relationship("Ciudades", back_populates="estados")


class Productos(Base):
    __tablename__ = "productos"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(11), nullable=False)
    descripcion = Column(String(25), nullable=False)
    categoria = Column(String(25), nullable=False)
    precio = Column(Integer, nullable=False)
    status = Column(Enum("y", "n"), nullable=False)

    pedidos = relationship("Pedidos", back_populates="producto")


class Ciudades(Base):
    __tablename__ = "ciudades"

    id = Column(Integer, primary_key=True, unique=True)
    id_estado = Column(
        ForeignKey("estados.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=False
    )
    nombre = Column(String(25), nullable=False)
    capital = Column(Integer, nullable=False)

    estados = relationship("Estados", back_populates="ciudades")
    clientes = relationship("Clientes", back_populates="ciudades")


class Clientes(Base):
    __tablename__ = "clientes"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(25), nullable=False)
    apellido = Column(String(25), nullable=False)
    codigo_postal = Column(
        ForeignKey("ciudades.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=False,
    )
    telefono = Column(String(11), nullable=False)

    ciudades = relationship("Ciudades", back_populates="clientes")
    pedidos = relationship("Pedidos", back_populates="cliente")


class Pedidos(Base):
    __tablename__ = "pedidos"

    id = Column(Integer, primary_key=True, unique=True)
    cliente_id = Column(ForeignKey("clientes.id"), nullable=False)
    fecha = Column(Date, nullable=False)
    producto_id = Column(ForeignKey("productos.id"), nullable=False)
    status = Column(Enum("y", "n"), nullable=False)

    cliente = relationship("Clientes", back_populates="pedidos")
    producto = relationship("Productos", back_populates="pedidos")

Este es el código fuente de modelos SQLAlchemy generado en base a la base de datos sistema.db. Más adelante se mejorara esta código fuente para obtener una mejor representación de los objetos.

2.9.5. Práctica - Caso real

A continuación se presenta una práctica más real de implementar el uso de proyectos con sqlacodegen, a continuación la estructura de proyecto llamado sistema:

proyectos/
└── sqlacodegen/
    └── sistema/
        ├── .env.example
        ├── __init__.py
        ├── main.py
        ├── models.py
        ├── requirements.txt
        └── settings.py

A continuación se presenta y explica el uso de cada archivo para este proyecto:

Archivo .env.example

Archivo plantilla dotenv del paquete adicional python-dotenv.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
### Common settings #############################
ENGINE_DB=sqlite
### SQLite #############################
DB=sistema.db
### MySQL/PostgreSQL #############################
# USER=
# PASSW=
# HOST=
# PORT=
# DB=

Archivo settings.py

Módulo de configuraciones del programa.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
"""Módulo de configuraciones del programa"""

import os
from dotenv import load_dotenv

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

load_dotenv()

ENGINE = os.getenv("ENGINE_DB")
USER = os.getenv("USER")
PASSW = os.getenv("PASSW")
HOST = os.getenv("HOST")
PORT = os.getenv("PORT")
DB = os.getenv("DB", "sistema.db")
DB_PATH = os.path.dirname(os.path.abspath(__file__)) + os.sep
DB_FILE = DB

engine = None

if ENGINE == "sqlite" and ENGINE is not None:
    if "DB_PATH" in globals() and "DB_FILE" in globals():
        # Conexión entre SQLAlchemy y SQLite3 DB API
        engine = create_engine(f"sqlite:///{DB_PATH}{DB_FILE}")
elif ENGINE == "mysql":
    if (
        "USER" in globals()
        and "PASSW" in globals()
        and "HOST" in globals()
        and "PORT" in globals()
        and "DB" in globals()
    ):
        # Conexión entre SQLAlchemy y MySQL DB API
        engine = create_engine(f"mysql+pymysql://{USER}:{PASSW}@{HOST}:{PORT}/{DB}")
elif ENGINE == "postgresql":
    if (
        "USER" in globals()
        and "PASSW" in globals()
        and "HOST" in globals()
        and "PORT" in globals()
        and "DB" in globals()
    ):
        # Conexión entre SQLAlchemy y PostgreSQL DB API
        engine = create_engine(f"postgresql://{USER}:{PASSW}@{HOST}:{PORT}/{DB}")
else:
    print(f"¡No se soporta ese tipo de conexión a base de datos {ENGINE}!")

if "engine" in globals():
    # Crear sesión con el engine de base de datos
    Session = sessionmaker(bind=engine)
    session = Session()
    # Crear base declarativa
    Base = declarative_base()

Archivo models.py

Módulo de modelos de SQLAlchemy.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
"""Módulo de modelos de SQLAlchemy"""

from settings import Base

from sqlalchemy import Column, Date, Enum, ForeignKey, Integer, String
from sqlalchemy.orm import relationship


class Estados(Base):
    """Clase de Estados

    Args:
        Base (Base): clase base declarativa

    Returns:
        class 'models.Estados': Clase de Estados
    """

    __tablename__ = "estados"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(25), nullable=False)
    codigo = Column(String(2), nullable=False)

    ciudades = relationship("Ciudades", back_populates="estados")

    def __repr__(self):
        return f"<Estados({self.nombre}, {self.codigo})>"

    def __str__(self):
        return f"Estado: {self.nombre}"


class Productos(Base):
    """Clase de Productos

    Args:
        Base (Base): clase base declarativa

    Returns:
        class 'models.Productos': Clase de Productos
    """

    __tablename__ = "productos"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(11), nullable=False)
    descripcion = Column(String(25), nullable=False)
    categoria = Column(String(25), nullable=False)
    precio = Column(Integer, nullable=False)
    status = Column(Enum("y", "n"), nullable=False)
    pedidos = relationship("Pedidos", back_populates="producto")

    def __repr__(self):
        return f"<Productos(nombre={self.nombre}, categoria={self.categoria}, precio={self.precio})>"

    def __str__(self):
        return f"Producto: {self.nombre} ({self.categoria}) - {self.precio:.2f}"


class Ciudades(Base):
    """Clase de Ciudades

    Args:
        Base (Base): clase base declarativa

    Returns:
        class 'models.Ciudades': Clase de Ciudades
    """

    __tablename__ = "ciudades"

    id = Column(Integer, primary_key=True, unique=True)
    id_estado = Column(
        ForeignKey("estados.id", ondelete="CASCADE", onupdate="CASCADE"), nullable=False
    )
    nombre = Column(String(25), nullable=False)
    capital = Column(Integer, nullable=False)
    estados = relationship("Estados", back_populates="ciudades")
    clientes = relationship("Clientes", back_populates="ciudades")

    def __repr__(self):
        return f"<Ciudades({self.nombre}, {self.estados})>"

    def __str__(self):
        return f"Ciudad: {self.nombre}, {self.estados}."


class Clientes(Base):
    """Clase de Clientes

    Args:
        Base (Base): clase base declarativa

    Returns:
        class 'models.Clientes': Clase de Clientes
    """

    __tablename__ = "clientes"

    id = Column(Integer, primary_key=True, unique=True)
    nombre = Column(String(25), nullable=False)
    apellido = Column(String(25), nullable=False)
    codigo_postal = Column(
        ForeignKey("ciudades.id", ondelete="CASCADE", onupdate="CASCADE"),
        nullable=False,
    )
    telefono = Column(String(11), nullable=False)

    ciudades = relationship("Ciudades", back_populates="clientes")
    pedidos = relationship("Pedidos", back_populates="cliente")

    def __repr__(self):
        return f"<Clientes({self.nombre} {self.apellido}, {self.telefono}, {self.ciudades}, {self.pedidos})>"

    def __str__(self):
        return f"Cliente: {self.nombre} {self.apellido}"


class Pedidos(Base):
    """Clase de Pedidos de ventas

    Args:
        Base (Base): clase base declarativa

    Returns:
        class 'models.Pedidos': Clase de Pedidos de ventas
    """

    __tablename__ = "pedidos"

    id = Column(Integer, primary_key=True, unique=True)
    cliente_id = Column(ForeignKey("clientes.id"), nullable=False)
    fecha = Column(Date, nullable=False)
    producto_id = Column(ForeignKey("productos.id"), nullable=False)
    status = Column(Enum("y", "n"), nullable=False)

    cliente = relationship("Clientes", back_populates="pedidos")
    producto = relationship("Productos", back_populates="pedidos")

    def __repr__(self):
        return f"<Pedidos({self.producto} {self.fecha}, {self.status})>"

    def __str__(self):
        return f"Pedido: {self.producto}, {self.fecha}"

Archivo main.py

Módulo principal del programa.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
"""Módulo principal del programa"""

import logging

from sqlalchemy import exc
from settings import DB_FILE, Base, engine, session
from models import Estados, Ciudades, Clientes, Productos, Pedidos

logging.basicConfig(level=logging.INFO)


def consulta_estados():
    """Consulta todos los estados"""

    print("✅ Lista de 10 Estados")
    # SELECT TOP 10 * FROM estados;
    todos_estados = session.query(Estados).limit(10).all()
    if len(todos_estados) == 0:
        logging.error("❌ ¡No hay ningún 'estado' con ese criterio en la base de datos!")
    else:
        row_count = 0
        for cada_estado in todos_estados:
            print(f"📜 {cada_estado}")
            row_count += 1
        logging.info(f"✅ ¡Consulta de los '{row_count}' estados!")


def consulta_ciudades():
    """Consulta todas las ciudades"""

    print("\n✅ Lista de 10 Ciudades")
    # SELECT TOP 10 * FROM ciudades;
    todos_ciudades = session.query(Ciudades).limit(10).all()
    if len(todos_ciudades) == 0:
        logging.error("❌ ¡No hay ningún 'ciudad' en la base de datos!")
    else:
        row_count = 0
        for cada_ciudad in todos_ciudades:
            print(f"📜 {cada_ciudad}")
            row_count += 1
        logging.info(f"✅ ¡Consulta de las '{row_count}' ciudades!")


def consulta_clientes():
    """Consulta todas las clientes"""

    print("\n✅ Lista de Clientes")
    # SELECT * FROM clientes;
    todos_clientes = session.query(Clientes).all()
    if len(todos_clientes) == 0:
        logging.error("❌ ¡No hay ningún 'cliente' en la base de datos!")
    else:
        row_count = 0
        for cada_cliente in todos_clientes:
            print(f"📜 {cada_cliente}")
            row_count += 1
        logging.info(f"✅ ¡Consulta de los '{row_count}' clientes!")


def consulta_productos():
    """Consulta todas las productos"""

    print("\n✅ Lista de Productos")
    # SELECT * FROM productos;
    todos_productos = session.query(Productos).all()
    if len(todos_productos) == 0:
        logging.error("❌ ¡No hay ningún 'producto' en la base de datos!")
    else:
        row_count = 0
        for cada_producto in todos_productos:
            print(f"📜 {cada_producto}")
            row_count += 1
        logging.info("✅ ¡Consulta de los '{row_count}' productos!")


def consulta_pedidos():
    """Consulta todas las pedidos"""

    print("\n✅ Lista de Pedidos")
    # SELECT * FROM pedidos;
    todos_pedidos = session.query(Pedidos).all()
    if len(todos_pedidos) == 0:
        logging.error("❌ ¡No hay ningún 'pedido' en la base de datos!\n")
    else:
        row_count = 0
        for cada_pedido in todos_pedidos:
            print(f"📜 {cada_pedido}")
            row_count += 1
        logging.info("✅ ¡Consulta de los '{row_count}' pedidos!")


if __name__ == "__main__":
    try:
        # Crea la base de datos y tablas
        Base.metadata.create_all(engine)
        logging.info(
            f"✅ ¡Creación exitosa de las tablas en la base de datos '{DB_FILE}'!\n"
        )
        # Consulta de productos
        consulta_estados()
        # Consulta de ciudades
        consulta_ciudades()
        # Consulta de clientes
        consulta_clientes()
        # Consulta de productos
        consulta_productos()
        # Consulta de pedidos
        consulta_pedidos()
    except exc.SQLAlchemyError as e:
        logging.error(
            f"❌ ERROR: ¡Se produjo un falla al establecer la conexión a la base de datos '{DB_FILE}': '{e}'!"
        )
    finally:
        if session:
            # Cerrar la conexión a la base de datos
            session.close()
            engine.dispose()
            logging.info(
                f"✅ ¡La conexión SQLite a la base de datos '{DB_FILE}' fue cerrada!"
            )

Archivo requirements.txt

Archivo de requirements.txt de la herramienta de gestión de paquetes pip.

1
2
3
4
# Install the packages from PyPi repository.
SQLAlchemy>=2.0.23 # For the last test date, version SQLAlchemy==2.0.38 was used.
python-dotenv==1.0.1
sqlacodegen==3.0.0

Teniendo creada la anterior estructura de proyecto, vuelva a ejecutar ahora el módulo con el siguiente comando, el cual a continuación se presentan el correspondiente comando de tu sistema operativo:

Antes de ejecutar debes instalar sus dependencias, con el siguiente comando:

pip3 install -r requirements.txt

Además debe crear el archivo .env en base a la plantilla .env.example y editarlo, con el siguiente comando:

cp .env.example .env && nano .env

Truco

El archivo .env se definen las configuraciones de conexión a la base de datos, puede modificarlo cambiar valores de la conexión.

Nota

Para conexiones a base de datos MySQL y PostgreSQL debe definir las variables que por defecto no están definidas.

Truco

Para ejecutar el código fuente de esta práctica debe invocar al módulo main.py, abra una consola de comando, acceda al directorio donde se encuentra la estructura previa y ejecute el siguiente comando:

python3 main.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Creación exitosa de las tablas en la base de datos 'sistema.db'!

✅ Lista de 10 Estados
📜 Estado: Amazonas
📜 Estado: Anzoátegui
📜 Estado: Apure
📜 Estado: Aragua
📜 Estado: Barinas
📜 Estado: Bolívar
📜 Estado: Carabobo
📜 Estado: Cojedes
📜 Estado: Delta Amacuro
📜 Estado: Falcón
INFO:root:✅ ¡Consulta de los '10' estados!

✅ Lista de 10 Ciudades
📜 Ciudad: Maroa, Estado: Amazonas.
📜 Ciudad: Puerto Ayacucho, Estado: Amazonas.
📜 Ciudad: San Fernando de Atabapo, Estado: Amazonas.
📜 Ciudad: Anaco, Estado: Anzoátegui.
📜 Ciudad: Aragua de Barcelona, Estado: Anzoátegui.
📜 Ciudad: Barcelona, Estado: Anzoátegui.
📜 Ciudad: Boca de Uchire, Estado: Anzoátegui.
📜 Ciudad: Cantaura, Estado: Anzoátegui.
📜 Ciudad: Clarines, Estado: Anzoátegui.
📜 Ciudad: El Chaparro, Estado: Anzoátegui.
INFO:root:✅ ¡Consulta de las '10' ciudades!

✅ Lista de Clientes
📜 Cliente: Leonardo Caballero
📜 Cliente: Ana Poleo
📜 Cliente: Manuel Matos
📜 Cliente: Liliana Andradez
INFO:root:✅ ¡Consulta de los '4' clientes!

✅ Lista de Productos
ERROR:root:❌ ¡No hay ningún 'producto' en la base de datos!

✅ Lista de Pedidos
ERROR:root:❌ ¡No hay ningún 'pedido' en la base de datos!

INFO:root:✅ ¡La conexión SQLite a la base de datos 'sistema.db' fue cerrada!

La ejecución anterior generar la siguiente estructura:

proyectos/
└── sqlacodegen/
    └── sistema/
        ├── __init__.py
        ├── .env
        ├── .env.example
        ├── main.py
        ├── models.py
        ├── requirements.txt
        ├── settings.py
        └── sistema.db

Archivo sistema.db

Archivo de base de datos de SQLite llamado sistema.db la cual no se incluye ya que cada vez que se inicia el programa main.py se elimina y crea nuevamente, para cuidar la creación de los datos iniciales.

Así de esta forma puede usar sqlacodegen para generar modelos SQLAlchemy desde una base de datos existente e implementar las operaciones ingresar, consultar, actualizar y eliminar registro en las tablas.

Ver también

Consulte la sección de lecturas suplementarias del entrenamiento para ampliar su conocimiento en esta temática.

¿Cómo puedo ayudar?

¡Mi soporte está aquí para ayudar!

Mi horario de oficina es de lunes a sábado, de 9 AM a 5 PM. UTM - Madrid, España.

La hora aquí es actualmente 7:35 PM UTM.

Mi objetivo es responder a todos los mensajes dentro de un día hábil.

Contáctenos en la sección de soporte
Contrata mi increíble soporte profesional