Python (XXVI) Sqlalchemy: Localizar e importar todas las clases a persistir. Modificar las tablas cuando la clase asociada cambia

1. Importar todas las clases a persistir con sqlalchemy

Para ello debemos tener todos los módulos que definen las clases dentro de una carpeta, en mi caso "models"

Creamos un módulo dentro de la carpeta "models" que recoja toda las clases y así solo tenemos que importar el array de clases a persistri.

El módulo se llama xmallmodels.py, y la variable que contendrá la clases es model_classes  y su código es:

import importlib
import pkgutil
import inspect
#------Imprescindible para poder importar de otras carpetas (de basicutils)
import sys
from pathlib import Path
path_root1 = Path(__file__).parents[1] # Damos un salto de directorio hacia arriba-> ximoutilsmod
sys.path.append(str(path_root1))
path_root0 = Path(__file__).parents[0] # El mismo directorio
sys.path.append(str(path_root0))
from basicutils.xmdb import Base
import models
# ------Fin imprescindible

model_classes = []

for loader, module_name, is_pkg in pkgutil.iter_modules(models.__path__):
    module = importlib.import_module(f"models.{module_name}")
    
    for name, obj in inspect.getmembers(module, inspect.isclass):
        # Ensure it's defined in the current module and is a subclass of Base (but not Base itself)
        if obj.__module__ == module.__name__ and issubclass(obj, Base) and obj is not Base:
            model_classes.append(obj)

# Example: register or print them
for cls in model_classes:
    print(f"Found model class: {cls.__name__} from {cls.__module__}")

Por ejemplo en nuestro programa de arranque, que intenta vincular los modelos con la BD, basta con importar model_classes desde este módulo. El código de este módulo mnu_main.py es:

#!/home/eduard/MyPython/11.softprop-01/venv_softprop/bin/python3

#1. Imports
from sqlalchemy import Table
#------Imprescindible para poder importar de otras carpetas (de basicutils)
import sys
from pathlib import Path
path_root1 = Path(__file__).parents[1] # Damos un salto de directorio hacia arriba-> ximoutilsmod
sys.path.append(str(path_root1))
path_root0 = Path(__file__).parents[0] # El mismo directorio
sys.path.append(str(path_root0))
from menus.mnu_fh import fh, app
#OJO: No eliminar las dependencias marcadas en gris (routes01mnu, routes02form, routes03grid,app) pues, sinó falla el programa
from menus import routes00comp, routes01mnu, routes02form, routes03grid # No eliminar !!!
from menus.mnu_fh import fh, app                         # No eliminar app !!!
from basicutils import xmdb
#--- Definición de tablas de la Base de datos postgres
from models.xmallmodels import model_classes
# ------Fin imprescindible

from basicutils import xmdb 

# Create database tables
xmdb.Base.metadata.create_all(bind=xmdb.engine)

# Execute the web server, but now not using fh.serve but uvicorm.run instead
#fh.serve()
if __name__ == "__main__":
	import uvicorn
	app.mount("/static", fh.StaticFiles(directory="/home/eduard/MyPython/11.softprop-01/static"), name="static")
	#uvicorn.run(app, host="0.0.0.0", port=5001, 
	#    ssl_keyfile="/home/eduard/MyPython/11.softprop-01/static/certs/wildcard2023Nginx.rsa", 
	#    ssl_certfile="/home/eduard/MyPython/11.softprop-01/static/certs/wildcard2023Nginx.crt")

	cert_path="/home/eduard/MyPython/11.softprop-01/static/certs/wildcard.municipio.es."
	#uvicorn.run(app, host="edu.tavernes.es", port=5001,	
	uvicorn.run(app, host="192.168.XXX.XXX", port=5001,	
		ssl_keyfile =cert_path+"key", 
		ssl_certfile=cert_path+"crt")

2. Modificar las tablas cuya clase asociada cambia

Para ello se buscan las clases que deriven de la clase Base, y si estan asociadas a una tabla (tienen el campo __tablename__) y se compara los atributos de la clase con las columnas de la tabla; y se añaden columnas  o se eliminan en base a los atributos de las clases.

Veamos el código del módulo xmdbupdate.py : 

''' Actualización de la estructura de tablas en función
     de las clases asociadas de SQLAlchemy.
'''

from sqlalchemy.sql import text
from sqlalchemy.inspection import inspect
from sqlalchemy.exc import ProgrammingError


#------Imprescindible para poder importar de otras carpetas (de basicutils)
import sys
from pathlib import Path
path_root1 = Path(__file__).parents[1] # Damos un salto de directorio hacia arriba-> ximoutilsmod
sys.path.append(str(path_root1))
path_root0 = Path(__file__).parents[0] # El mismo directorio
sys.path.append(str(path_root0))
#OJO: No eliminar las dependencias marcadas en gris
#--- Definición de tablas de la Base de datos postgresfrom basicutils.xmdb import Base 
from basicutils import xmdb
from models.xmallmodels import model_classes
# ------Fin imprescindible

#-----------------------------------------------------------
# --- Configuration ---
session = xmdb.session_local()

def all_table_subclasses(cls):
    """Recursively find all subclasses of a class that has a table assigned."""
    subclasses = set()

    for subclass in cls.__subclasses__():
        print (subclass.__name__,getattr(subclass, '__tablename__', ''), getattr(subclass, '__abstract__', False))
        if len(getattr(subclass, '__tablename__', '')) > 0:
            subclasses.add(subclass)
        subclasses.update(all_table_subclasses(subclass))

    return subclasses



# --- Update function ---
def update_table_structure(cls):
    table_name = cls.__tablename__
    table_args = cls.__table_args__
    schema = table_args.get('schema') if isinstance(table_args, dict) else None
    inspector = inspect(xmdb.engine)

    if not inspector.has_table(table_name, schema=schema):
        print(f"Table '{schema}.{table_name}' does not exist. Creating it...")
        cls.__table__.create(xmdb.engine)
        return

    db_columns = {col["name"]: col for col in inspector.get_columns(table_name, schema=schema)}
    model_columns = {col.name: col for col in cls.__table__.columns}

    #Execute a transaction
    with xmdb.engine.begin() as conn:
        # Add missing columns
        for name, column in model_columns.items():
            if name not in db_columns:
                col_type = column.type.compile(xmdb.engine.dialect)
                nullable = "NULL" if column.nullable else "NOT NULL"
                alter = f'ALTER TABLE "{schema}"."{table_name}" ADD COLUMN "{name}" {col_type} {nullable};'
                print("Adding:", alter)
                conn.execute(text(alter))

        # Drop extra columns
        for name in db_columns:
            if name not in model_columns:
                alter = f'ALTER TABLE "{schema}"."{table_name}" DROP COLUMN "{name}";'
                print("Dropping:", alter)
                try:
                    conn.execute(text(alter))
                except ProgrammingError as e:
                    print(f"Could not drop column {name}: {e}")

# --- Run for all subclasses of Base for updating the tables---
def update_all_tables(base):
    for cls in all_table_subclasses(base):
        update_table_structure(cls)

if __name__ == "__main__":
	update_all_tables(xmdb.Base)
	session.close()
	xmdb.engine.dispose()