SqlAlchemy (I) : Conexión, crer DB, Tablas y columnas. Ejecutar SQL a pelo

Veamos este código python (db.py).

OJO: Para ejecutar SQL "a pelo" o raw SQL hay que ver esta entrada, pues nos llevaremos alguna sorpresa

import sqlalchemy as db
import yaml

#1. Read data from the static/tables folder and Create tables
#1.a Read the tables information from file
def getTables(tablesFilePath: str='static/tables/prov.yml')->list:
    with open(tablesFilePath) as f: 
        tableDict=yaml.safe_load(f)	
    return tableDict

#1.b Create Columns of the Table
def getColumn(col: list)->db.Column:
    lst=[]
    aDict={}
    for i, elem in enumerate(col):
        if i==0: lst.append(elem)
        elif i==1: 
            sublst=elem.split('(')
            type =sublst[0]
            lenOrFkey='255'
            if len(sublst)>1:
                lenOrFkey=sublst[1].replace(')','')
            match type.lower():
                case 'int' | 'integer': lst.append(db.Integer())
                case 'float': lst.append(db.Float)
                case 'bool': lst.append(db.Boolean)
                case 'str' | 'string' | 'char' | 'varchar' : lst.append(db.String(int(lenOrFkey))) 
                case 'date': lst.append(db.Date())
                case 'timestamp': lst.append(db.TIMESTAMP())
                case 'json': lst.append(db.JSON())
                case 'foreignkey' : lst.append(db.ForeignKey(lenOrFkey))
    else: aDict |= elem
    column=db.Column(*lst, **aDict)    
    print (str(column))
    return column          
                
#1.c Create the tables from the information                
def getTable(table: dict, metadataObj: db.MetaData)->db.Table:
    return db.Table(table['name'], metadataObj, *[getColumn(col) for col in table['columns']])   


#2. Insert a list of elements in a table
#2.1 Using sqlAlchemist
def insert(table: db.Table | str, listDataDict: list[dict],conn: db.engine.base.Connection=None, commit:bool=False):
    """
    Insert a list of dictionaries in the given table.
    
    Parameters
    ----------
    table: db.Table | str
        Either a db.Table object or the name of the table
    listDataDict: list[dict]
        A list of dictionaries where each dictionary represents a row to be inserted
    conn: db.engine.base.Connection, optional
        The connection to use for the insert. If None, it will use the default engine.
    commit: bool, optional
        If True, will commit the transaction. Defaults to False.
    
    
    Returns
    -------
    A list of tuples of ids of the inserted rows.
    """
    mytable=db.Table('Student', metadataObj, autoload_with=conn.engine) if isinstance(table, str) else table
    query = db.insert(mytable).values(listDataDict).returning(table.c.id)
    Result = conn.execute(query)
    result=Result.fetchall()
    if commit: conn.commit()
    return result

#2.1 Using raw SQL
def insertSQL(table: str, listDataDict: list[dict],conn: db.engine.base.Connection=None, commit:bool=False):
    for aDict in listDataDict:
        fields='(' + ','.join(str(key) for key in aDict.keys()) + ')'
        values=fields.replace("(", "( :"). replace(",",", :") 
        command="INSERT INTO "+ table + fields + "VALUES "+ values 
        conn.execute(db.text(command), **aDict)
        if commit: conn.commit()
        
    
    
#3. List all elements in a table
def listAll(table: db.Table | str, conn: db.engine.base.Connection=None):
    mytable=db.Table('Student', metadataObj, autoload_with=conn.engine) if isinstance(table, str) else table
    query = db.select([mytable])
    result = conn.execute(query)
    return result

#3.1 Execute a command from SQL 
def execSQL(sql: str, conn: db.engine.base.Connection=None, commit: bool=False) ->list:   
    result=conn.execute(db.text(sql))  
    if commit: conn.commit()
    else: return result.fetchall()
    
            
engine=db.create_engine('sqlite:///test.db', echo=True)
conn=engine.connect()

metadataObj = db.MetaData() #extracting the metadata


#4. Construct the DB
'''
tableLst=getTables(tablesFilePath='static/tables/prov.yml')
dbTableLst=[]
for tableInfo in tableLst:
    table=getTable(tableInfo, metadataObj)
    print(repr(table))
    dbTableLst.append(table)
'''
sql="INSERT INTO Student (name, age, tutor) VALUES('Peret',14,1)"
execSQL(sql,conn,True)

sql="CREATE VIEW VWKK AS SELECT * FROM Student a join Teacher b on a.tutor=b.id"
execSQL(sql,conn,True)

result=execSQL('SELECT * FROM VWKK',conn)
#tuples=result.fetchall()
for t in result:
    print(str(t))
    
    
'''
query = db.insert(dbTableLst[0]).values(name='Matthew', subject="Quantum Mechanics").returning(dbTableLst[0].c.id)
Result = conn.execute(query)
print(Result.fetchone().id)   
query = db.insert(dbTableLst[0]).values(name='Motos', subject="Ampliació de matemàtiques").returning(dbTableLst[0].c.id)
Result = conn.execute(query)
print(Result.fetchone().id)   

query = db.insert(dbTableLst[1]).values(name='Matthew', age=18, tutor=1).returning(dbTableLst[1].c.id)
Result = conn.execute(query)
print(Result.fetchone().id)   
query = db.insert(dbTableLst[1]).values(name='Pepet', age=18, tutor=2).returning(dbTableLst[1].c.id)
Result = conn.execute(query)
print(Result.fetchone().id)   
conn.commit()

output = conn.execute(dbTableLst[0].select()).fetchall()
print(output)
output = conn.execute(dbTableLst[1].select()).fetchall()
print(output)

Student= db.Table('Student', metadataObj, autoload_with=engine) #Table object

#Teacher= db.Table('Teacher', metadata, autoload_with=engine) #Table object
#print(repr(metadata.tables['Student','Teacher']))

    
'''    

Esta es la configuración de las tablas en formato yml (static/tables/prov.yml)

 - name: Teacher
   columns:
      - [id, int, primary_key: true, autoincrement: true]
      - [name, str(50), nullable : false]
      - ['subject', str(50), default: Maths]


 - name: Student
   columns:
      - [id, int, primary_key: true, autoincrement: true]
      - [name, str(50), nullable : false]
      - [age, int, default: 15]
      - [tutor, ForeignKey(Teacher.id), nullable: false]