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()
