ERREUR <Ens>ErrCanNotAcquireJobRootLock avec sqlalchemy et embedded python

Bonjour,

Je ne peux pas reproduire votre erreur. Il me manque quelques informations.

Ce que j'ai fait jusqu'à présent est :

• ajouter les imports manquants
• ajouter la classe DFrameRequest manquante
  • je suppose que c'est une dataclass avec un champ nommé dframe de type pd.DataFrame
  • je suppose que c'est une sous-classe de Message
• j'ai ajouté une fonction main pour exécuter le code
  • je ne suis pas sûr du format du dataframe et des données qu'il contient

from dataclasses import dataclass
import pandas as pd
from grongier.pex import BusinessOperation,Message
from sqlalchemy import create_engine, types

@dataclass
class DFrameRequest(Message):
    dframe: pd.DataFrame

class FileOperationEmbedded(BusinessOperation):
    tablename = None
    engine = None

    def on_init(self):
        if not hasattr(self, "dsnIris"):
            self.dnsIris = 'iris+emb:///'
        if not hasattr(self, "schema"):
            self.schema = 'Toto'

        self.engine = create_engine(self.dnsIris)
        return None

    def on_message(self, request:DFrameRequest):
        df = pd.DataFrame(request.dframe.col)

        for row in request.dframe.col:
            df = pd.DataFrame.from_dict(row, orient='index').T.reset_index(drop=True)
            try:
                df.to_sql(name=self.tablename, con=self.engine, if_exists='append', index=False, schema=self.schema,
                          dtype={'id': types.INTEGER, 'col_type': types.VARCHAR(50), 'col_center': types.VARCHAR(50),
                                 'col_name': types.VARCHAR(50), 'col_issue_name': types.VARCHAR(50),
                                 'col_model': types.VARCHAR(50), 'col_treatment': types.VARCHAR(50),
                                 'source': types.VARCHAR(50), 'filename': types.VARCHAR(100), 'created_at': types.TIMESTAMP})
            except Exception as e:
                self.log_info(f"Une erreur s'est produite : {e}")

        return None
    
if __name__ == '__main__':
    # create a new instance of the business operation
    bo = FileOperationEmbedded()
    # initialize the business operation
    bo.on_init()

    # create a new message
    msg = DFrameRequest(pd.DataFrame())
    msg.dframe.col = [
            {'id': 1, 'col_type': 'type1', 'col_center': 'center1', 'col_name': 'name1', 'col_issue_name': 'issue1',
             'col_model': 'model1', 'col_treatment': 'treatment1', 'source': 'source1', 'filename': 'file1',
             'created_at': '2021-10-01 00:00:00'},
            {'id': 2, 'col_type': 'type2', 'col_center': 'center2', 'col_name': 'name2', 'col_issue_name': 'issue2',
             'col_model': 'model2', 'col_treatment': 'treatment2', 'source': 'source2', 'filename': 'file2',
             'created_at': '2021-10-02 00:00:00'}
        ]

    # send the message to the business operation
    bo.on_message(msg)
    print("Done")

Ensuite, à partir de votre code, je peux voir les problèmes suivants :

• vous utilisez le même nom de variable pour le dataframe et la liste des lignes
• la variable self.tablename n'est pas initialisée
• le nom FileOperationEmbedded n'est peut-être pas le meilleur nom pour votre classe car ce n'est pas une opération de fichier
• pourquoi utilisez-vous une boucle for pour itérer sur les lignes du dataframe ?

J'ai modifié votre code pour corriger ces problèmes :

from dataclasses import dataclass
import pandas as pd
from grongier.pex import BusinessOperation,Message
from sqlalchemy import create_engine, types

@dataclass
class DFrameRequest(Message):
    dframe: pd.DataFrame

class IrisSqlAlchmyEmbedded(BusinessOperation):
    tablename = None
    engine = None

    def on_init(self):
        if not hasattr(self, "dsnIris"):
            self.dnsIris = 'iris+emb:///'
        if not hasattr(self, "schema"):
            self.schema = 'Toto'
        if not hasattr(self, "tablename") or self.tablename is None:
            self.tablename = 'mytable'

        self.engine = create_engine(self.dnsIris)
        return None

    def on_message(self, request:DFrameRequest):

        try:
            request.dframe.to_sql(name=self.tablename, con=self.engine, if_exists='append', index=False, schema=self.schema,
                      dtype={'id': types.INTEGER, 'col_type': types.VARCHAR(50), 'col_center': types.VARCHAR(50),
                             'col_name': types.VARCHAR(50), 'col_issue_name': types.VARCHAR(50),
                             'col_model': types.VARCHAR(50), 'col_treatment': types.VARCHAR(50),
                             'source': types.VARCHAR(50), 'filename': types.VARCHAR(100), 'created_at': types.TIMESTAMP})
        except Exception as e:
            print(f"Une erreur s'est produite : {e}")

        return None

if __name__ == '__main__':
    # create a new instance of the business operation
    bo = IrisSqlAlchmyEmbedded()
    # initialize the business operation
    bo.on_init()

    # create a new message
    msg = DFrameRequest(pd.DataFrame([
            {'id': 1, 'col_type': 'type1', 'col_center': 'center1', 'col_name': 'name1', 'col_issue_name': 'issue1',
             'col_model': 'model1', 'col_treatment': 'treatment1', 'source': 'source1', 'filename': 'file1',
             'created_at': '2021-10-01 00:00:00'},
            {'id': 2, 'col_type': 'type2', 'col_center': 'center2', 'col_name': 'name2', 'col_issue_name': 'issue2',
             'col_model': 'model2', 'col_treatment': 'treatment2', 'source': 'source2', 'filename': 'file2',
             'created_at': '2021-10-02 00:00:00'}
        ]))

    # send the message to the business operation
    bo.on_message(msg)
    print("Done")