Discussion :

[dedalios]

Code de lancement :lance.py

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
# coding: utf-8  
'''
Created on 23 mars 2018
 
@author: 
'''
import csv
 
import sys
import csv
import os
import encodings
from collections import namedtuple
 
 
import psycopg2                                # Accés PostgreSQL
import psycopg2.extras                        # Goodies PostgreSQL
import sys                                    # Parametres et fonctions systèmes
import os                                    # Gestion os
 
from  src.Table_VM  import *                       # creation des Tables triboc
from src.FIchier_Vm.src.convertie import *
 
class latable():
    def __init__(self):
 
        self.table ='public.vsl_actes'  
        self.base =''  
        self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %(ACT_CODE)s , %(ACT_CODE)s );'''    
 
        self.selection =  'select * from  public.vsl_actes'   
 
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = 'NULL'
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = 'NULL'
        return super().__new__(cls, *row)
 
 
class Writer():
    def __init__(self):
        global base_vm
 
        self.curseur = base_vm.connect.cursor()
        self.matable =latable()
 
    def writesmany(self, records):
 
        rq =self.matable.insertion
 
        with self.curseur:
            print(records)
            try:
                self.curseur.executemany(rq, records)
            except psycopg2.DataError as e:
                print("Problème"+ " :" +   e.args[0]  + " sur commande d'insertion /n" )  
 
                print("\t"  +str(records))
                self.curseur.rollback()
                # dans le cas d'un problème d'injection multipls on repasse à la mono injection au cas par cas
                # et on laisse les mécasnisme de sécurite de la base agir
                self.writemono( records )
            else: 
                self.curseur._cnx.commit()
 
    def writemono(self, records):
        matable =latable()
        rq =matable.insertion
 
 
        with self.curseur:
            for enregistrement in records:
                try:
                    self.curseur.execute(rq, enregistrement)
                except psycopg2.DataError as e:
                    print("Problème"+ " :" +   e.args[0]  + " sur commande d'insertion /n" )  
 
                    print("\t"  +str(enregistrement))
                    self.curseur.rollback()
                else: 
                    self.curseur.commit()
 
    def dump(self):
        matable =latable()
        rq =matable.selection
        with self._cnx:
            for row in self.curseur.execute(rq):
                print (row)
 
class vsl_acte(object):
    def __init__(self, nomfichier ):
        global base_vm
        self.nomfichier = nomfichier
 
        print('Injection ' + self.nomfichier +' sur la base ' + base_vm.dbname )
 
        writer = Writer()
        with open(self.nomfichier   , encoding='utf-8') as file:
            reader = csv.reader(file,'excel-fr')
            reader.__next__() # bypass l'entete du fichier
            packer = Packer(reader, record)
            i = 1
 
            for records in packer.get_records():
                writer.writesmany(records)
                #print(i, records)
                i += 1         
 
if __name__ == '__main__':
    from src.Explorateur import *
    encodage ="utf-8"
 
    csv.register_dialect('excel-fr', ExcelFr())
    rep = explorateur_dir('.//') 
    if  rep=='':
        print ("Pas de répertoire")
        exit(1)
    else:
        print ("vous avez choisi le repertoire : "+ rep)
 
        fichier_csv = rep +'/' + 'actes.csv'      
        BaseSQL = 'RecetteSTA'
 
 
        # Reconstution de la base
        base_vm = Creation_table_VM()
        base_vm.vsl_actes() # creation de la base de données
 
        vsl_acte(fichier_csv ) #

Table_VM.py

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
# coding: utf-8  
import psycopg2                                # Acc�s PostgreSQL
import psycopg2.extras                        # Goodies PostgreSQL
import sys, getopt, os, time
 
class Creation_table_VM(object):
 
    """ Structure des tables  """
 
    def __init__(self):
        self.dbname='MaBase'
        self.user = 'TOTO'
        self.password = 'MONCODE'
        self.host ='localhost'
 
        ref_selfect = "dbname=" + self.dbname + " user=" + self.user + " password=" +self.password  + " host=" +self.host
        try:
            self.connect = psycopg2.connect(ref_selfect)
        except psycopg2.Error as e:
            print("Connection base impossible!")
            print(e.pgerror)
            print(e.diag.message_detail)
            sys.exit(1)
 
    def execute_requete(self,requete):
 
        cur = self.connect.cursor()
 
        try:
            cur.execute(requete)
        except psycopg2.Error as e:
            print("Problème sur traitement de la requête: \n")
            print(requete + '\n')
            print(e.pgerror)
            print(e.diag.message_detail)
            sys.exit(1)
        else:
            self.connect.commit()
            cur.close()
 
    def execute_insert(self,requete):
        cur = self.connect.cursor()        
 
        cur.close()
 
    def close(self):
        self.connect.close()    
 
    def Genere_Base(self):
         #Table Primere 
        self.vsl_actes()   
 
 
    def vsl_actes(self): 
 
        drop_tbl = "DROP TABLE if exists public.vsl_actes CASCADE;"
        self.execute_requete(drop_tbl)
 
        create_tbl = """CREATE TABLE public.vsl_actes (
            act_code varchar(255) NOT NULL,
            act_libelle varchar(255) NULL,
            CONSTRAINT vsl_actes_pk PRIMARY KEY (act_code)
        )
        WITH (
            OIDS=FALSE
        ) ;"""
        self.execute_requete(create_tbl)
 
        create_index = "CREATE INDEX vsl_actes_idx ON public.vsl_actes USING btree (act_libelle) ;"
        self.execute_requete(create_index)

convertie.py

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
# coding: utf-8  
'''
Created on 23 mars 2018
 
@author: 
'''
import csv
from datetime import datetime
from datetime import date
 
def convertie_date(src):
    if src =='NULL':
        valid_date = None
    else:
        if src =='99991231':
            valid_date = None
        else:
 
            try:
                alpha_aaaa = int(src[0:4]) 
            except ValueError:
                print(src)
            alpha_mm =   int(src[4:6]) 
            alpha_jj =  int(src[6:8]) 
            try :
                valid_date = date(alpha_aaaa,alpha_mm,alpha_jj) 
                return valid_date
            except ValueError:
                print('%s Date invalide !' % src)
                return None
pass  
 
 
 
def convertie_date_a(src):
    if src =='NULL':
        valid_date = None
    else:
        alpha_aaaa = int(src[0:4]) 
        alpha_mm =   int(src[4:6]) 
        alpha_jj =  int(src[6:8]) 
        try :
            valid_date = date(alpha_aaaa,alpha_mm,alpha_jj) 
            return valid_date
        except ValueError:
            print('%s Date invalide !' % src)
            return None
pass  
 
 
 
def EstceNull(src):
    if src =='NULL':
        return  None
    else:
        return src
 
 
def Estce_nombreNull(src):
    if src =='NULL':
        return  None
    else:
        return int(src)    
 
 
class ExcelFr(csv.excel):
    # Séparateur de champ
    delimiter = ";"
    # Séparateur de ''chaîne''
    quotechar = None
    # Gestion du séparateur dans les ''chaînes''
    escapechar = None
    doublequote = True
    # Fin de ligne
    lineterminator = "\r\n"
    # Ajout automatique du séparateur de chaîne (pour ''writer'')
    quoting = csv.QUOTE_NONE
    # Ne pas ignorer les espaces entre le délimiteur de chaîne
    # et le texte
    skipinitialspace = False 
 
 
 
 
class Packer:
    def __init__(self, iterable, factory):
        self._iterable = iterable
        self._factory = factory
 
    def get_records(self, count=500):
        done = False
        while not done:
            records = []
            for _ in range(count):
                try:
                    row = next(self._iterable)
                except StopIteration:
                    done = True
                    break
                except:
                    raise
                #enregistrement = self._factory(row)
                #if enregistrement not in records:
                records.append(self._factory(row))
            yield records
        raise StopIteration

J'ai un problème lors de l'insertion des données dans POSTGRE

Dans cette partie du code

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
    def writesmany(self, records):
 
        rq =self.matable.insertion
 
        with self.curseur:
            print(records)
            try:
                self.curseur.executemany(rq, records)
            except psycopg2.DataError as e:
                print("Problème"+ " :" +   e.args[0]  + " sur commande d'insertion /n" )  
 
                print("\t"  +str(records))
                self.curseur.rollback()
                # dans le cas d'un problème d'injection multipls on repasse à la mono injection au cas par cas
                # et on laisse les mécasnisme de sécurite de la base agir
                self.writemono( records )
            else: 
                self.curseur._cnx.commit()

Le print retourne ceci
[record(ACT_CODE='MON', ACT_LIBELLE='Monture'), record(ACT_CODE='VER', ACT_LIBELLE='VERRE'), ....]


mais problème de l'insert

self.curseur.executemany(rq, records)
TypeError: tuple indices must be integers or slices, not str

Je ne trouve pas la bonne syntaxe pour l'insert

Cette ligne d'insertion ne semble pas accepter

INSERT INTO public.vsl_actes
(ACT_CODE, ACT_LIBELLE)
VALUES ( %(ACT_CODE)s , %(ACT_CODE)s );

comment puis-je transmettre dans VALUES les enregistrements concernés ?

[dedalios]

Je viens de trouver en partie la solution
le lanceur

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
# coding: utf-8  
'''
Created on 23 mars 2018

@author: 
'''
import csv
 
import sys
import csv
import os
import encodings
from collections import namedtuple


import psycopg2                                # Accés PostgreSQL
import psycopg2.extras                        # Goodies PostgreSQL
import sys                                    # Parametres et fonctions systèmes
import os                                    # Gestion os


from  src.Table_TRB  import *                       # creation des Tables triboc
from  src.Table_VM  import *                       # creation des Tables triboc

from src.FIchier_Vm.src.convertie import *

class latable():
    def __init__(self):

        self.table ='public.vsl_actes'  
        self.base =''  
        self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %s , %s );'''    
 
        self.selection =  'select * from  public.vsl_actes'   

class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = 'NULL'
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = 'NULL'
        return super().__new__(cls, *row)
 
  
class Writer():
    def __init__(self):
        global base_vm
 
        self.curseur = base_vm.connect.cursor()
        self.matable =latable()

    def writesmany(self, records):
        self.curseur = base_vm.connect.cursor()
        rq =self.matable.insertion

        with self.curseur:
            print(records)
            try:
                self.curseur.executemany(rq, records)
            except psycopg2.DataError as e:
                print("Problème"+ " :" +   e.args[0]  + " sur commande d'insertion /n" )  
 
                print("\t"  +str(records))
                base_vm.rollback()
                # dans le cas d'un problème d'injection multipls on repasse à la mono injection au cas par cas
                # et on laisse les mécasnisme de sécurite de la base agir
                self.writemono( records )
            else: 
               base_vm.commit()

                
    def writemono(self, records):
        matable =latable()
        rq =matable.insertion
        
        
        with self.curseur:
            for enregistrement in records:
                try:
                    self.curseur.execute(rq, enregistrement)
                except psycopg2.DataError as e:
                    print("Problème"+ " :" +   e.args[0]  + " sur commande d'insertion /n" )  
     
                    print("\t"  +str(enregistrement))
                    self.curseur.rollback()
                else: 
                    self.curseur.commit()
 
    def dump(self):
        matable =latable()
        rq =matable.selection
        with self._cnx:
            for row in self.curseur.execute(rq):
                print (row)

   


class vsl_acte(object):
    def __init__(self, nomfichier ):
        global base_vm
        self.nomfichier = nomfichier
 
        print('Injection ' + self.nomfichier +' sur la base ' + base_vm.dbname )
        
        writer = Writer()
        with open(self.nomfichier   , encoding='utf-8') as file:
            reader = csv.reader(file,'excel-fr')
            reader.__next__() # bypass l'entete du fichier
            packer = Packer(reader, record)
            i = 1
            
            for records in packer.get_records():
                writer.writesmany(records)
                #print(i, records)
                i += 1         

if __name__ == '__main__':
    from src.Explorateur import *
    encodage ="utf-8"
    
    csv.register_dialect('excel-fr', ExcelFr())
    rep = explorateur_dir('.//') 
    if  rep=='':
        print ("Pas de répertoire")
        exit(1)
    else:
        print ("vous avez choisi le repertoire : "+ rep)
        
        fichier_csv = rep +'/' + 'actes.csv'      
        BaseSQL = 'RecetteSTA'
        
        
        # Reconstution de la base
        base_vm = Creation_table_VM()
        base_vm.vsl_actes() # creation de la base de données
       
        vsl_acte(fichier_csv ) # 
        base_vm.close()

modfication de Table_VM.py

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
# coding: utf-8  
import psycopg2                                # Acc�s PostgreSQL
import psycopg2.extras                        # Goodies PostgreSQL
import sys, getopt, os, time
 
class Creation_table_VM(object):
 
    """ Structure des tables  """
 
    def __init__(self):
        self.dbname='MaBase'
        self.user = 'TOTO'
        self.password = 'MONCODE'
        self.host ='localhost'
 
        ref_selfect = "dbname=" + self.dbname + " user=" + self.user + " password=" +self.password  + " host=" +self.host
        try:
            self.connect = psycopg2.connect(ref_selfect)
        except psycopg2.Error as e:
            print("Connection base impossible!")
            print(e.pgerror)
            print(e.diag.message_detail)
            sys.exit(1)
 
    def execute_requete(self,requete):
 
        cur = self.connect.cursor()
 
        try:
            cur.execute(requete)
        except psycopg2.Error as e:
            print("Problème sur traitement de la requête: \n")
            print(requete + '\n')
            print(e.pgerror)
            print(e.diag.message_detail)
            sys.exit(1)
        else:
            self.connect.commit()
            cur.close()
 
 
 
    def close(self):
        self.connect.close()    
 
    def commit(self):
        self.connect.commit()    
 
 
    def rollback(self):
        self.connect.rollback()        
 
    def Genere_Base(self):
         #Table Primere 
        self.vsl_actes()   
 
 
    def vsl_actes(self): 
 
        drop_tbl = "DROP TABLE if exists public.vsl_actes CASCADE;"
        self.execute_requete(drop_tbl)
 
        create_tbl = """CREATE TABLE public.vsl_actes (
            act_code varchar(255) NOT NULL,
            act_libelle varchar(255) NULL,
            CONSTRAINT vsl_actes_pk PRIMARY KEY (act_code)
        )
        WITH (
            OIDS=FALSE
        ) ;"""
        self.execute_requete(create_tbl)
 
        create_index = "CREATE INDEX vsl_actes_idx ON public.vsl_actes USING btree (act_libelle) ;"
        self.execute_requete(create_index)

[wiztricks]

Salut,

Quand vous avez ce genre de problème, il faut commencer par essayer de le reproduire avec un script minimal (qui contient juste le nécessaire pour...).

D'autant qu'une lecture diagonale fait tiquer sur:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
       self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %(ACT_CODE)s , %(ACT_CODE)s );''' 
...
        self.curseur.executemany(rq, records)

ou:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = 'NULL'
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = 'NULL'
        return super().__new__(cls, *row)

car cela ne fera probablement pas ce que vous en attendez.
Dit autrement, coder c'est aussi l'assemblage de constructions qu'on maîtrise: on a écrit des petits scripts qui permettent de vérifier que çà fait bien ce qu'on en attend.
A défaut, si çà ne veut plus marcher, on ne sait plus où donner de la tête (et on se retrouve à écrire des scripts qui assurent que les "bouts" fonctionnent)...

- W

[dedalios]

Salut,

Quand vous avez ce genre de problème, il faut commencer par essayer de le reproduire avec un script minimal (qui contient juste le nécessaire pour...).

D'autant qu'une lecture diagonale fait tiquer sur:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
       self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %(ACT_CODE)s , %(ACT_CODE)s );''' 
...
        self.curseur.executemany(rq, records)

ou:

- W

Bonjour lorsque j'ai recherche des scripts utilisant psycopg2 et dans 90% des exemples on retrouve se type de codage pour les insert postgre , pour autant le second post rédigé j'ai mise la correction de l'appel à la requête .

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
 self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %s , %;'''

avoir même

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
        self.insertion =  '''INSERT INTO public.vsl_actes
        (ACT_CODE, ACT_LIBELLE) 
        VALUES (  %s , %s )
        ON CONFLICT (ACT_CODE) DO NOTHING;'''

pour l'éviter de me prendre la tête avec des personnes qui ne fournissent pas de données propre a insérer , présence de doublon ... etc

Autre point même si psycopg2 est utilisant avec Python3.x les scripts sont essentiellement orienté Python2.x ce qui quelque fois demande des adaptations.

pour ce qui concerne l'autre code

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
 
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = 'NULL'
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = 'NULL'
        return super().__new__(cls, *row)

oui effectivement est porteur d'une erreur , d'un non sans que je n'avais pas remarqués

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = None
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = None
        return super().__new__(cls, *row)

[wiztricks]

Bonjour lorsque j'ai recherche des scripts utilisant psycopg2 et dans 90% des exemples on retrouve se type de codage pour les insert postgre

Les exemples que vous trouvez sur Internet sont une chose. Écrire un code sans le tester en est une autre (et c'est pas parce que vous récupérez un truc sur Internet qu'il est "correct" ou "adapté" à vos besoins).

oui effectivement est porteur d'une erreur , d'un non sans que je n'avais pas remarqués

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
    def __new__(cls, row):
        if row[0] == 'NULL':
            record.ACT_CODE = None
        if   row[1] == 'NULL': 
             record.ACT_LIBELLE = None
        return super().__new__(cls, *row)

C'est pas si compliqué d'imaginer comment tester que çà fait bien ce que vous attendez.... Mais c'est sûr que "tester" une construction n'est pas quelque chose que vous allez trouver sur Internet.

- W

[fred1599]

Code de lancement :lance.py

Mon code de lancement c'est Bonjour, ça peut paraître dérisoire, mais c'est important pour débuter sous de bonnes résolutions la lecture de votre prose (comme dirait wiztricks ).

Qui puis est vous ne parlez qu'avec du code d'une certaine longueur (trop importante) avec beaucoup d'erreurs, hors votre problème ne devrait comporter que quelques lignes représentant cette erreur. Ça fait quand même pas mal de fois que la remarque est faîte. Je voudrai pas insister, mais bon... un moment donné, on attend un peu d'efforts de votre part.

Donc si vous pouviez poser comme l'indique wiztricks un code minimaliste qui reproduit cette erreur, je suis quasi certains que l'erreur sera résolue dans les deux-trois réponses qui suivront votre demande.

Merci

[dedalios]

Bonjour
si vous faite la supposition que je ne test pas le code avant de le poster vous faites erreur.

[wiztricks]

si vous faite la supposition que je ne test pas le code avant de le poster vous faites erreur.

Si c'était le cas, pourquoi ne pas poster ces scripts de tests plutôt qu'un programme complet ...

- W

[dedalios]

Si c'était le cas, pourquoi ne pas poster ces scripts de tests plutôt qu'un programme complet ...

- W

parce cette semble de script est le test de base sur le quelle s'appuie l'application globale faite de plus 30 de script .

[wiztricks]

parce cette semble de script est le test de base sur le quelle s'appuie l'application globale faite de plus 30 de script .

C'est ce qu'on pourrait appeler des tests fonctionnels. Avant il y a les tests unitaires, qui s'assurent que les petits bouts font ce qu'on en attend avant de les utiliser.
Relisez ce code là:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
>>> from collections import namedtuple
>>> class record(namedtuple('record', 'ACT_CODE, ACT_LIBELLE')):
...     def __new__(cls, row):
...         if row[0] == 'NULL':
...             record.ACT_CODE = None
...         if   row[1] == 'NULL':
...              record.ACT_LIBELLE = None
...         return super().__new__(cls, *row)
...

On peut supposer que l'intention est de remplacer les "NULL" des deux premières colonnes par None - sinon je ne vois pas trop à quoi cela pourrait servir.
Bon, çà se teste:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
>>> record(['NULL', None])
record(ACT_CODE='NULL', ACT_LIBELLE=None)
>>>

Donc vous avez écrit un code qui pourrait avoir l'intention de... mais hélas, comme vous n'avez pas pris le temps de tester, çà ne le fait pas... Et çà foutra la grouille plus tard.
Et si vous accumulez la composition de constructions non testées, je comprends que ce soit prise de tête à mettre au point...

Pour autant vous puissiez espérer de l'aide pour expliquer pourquoi "record" ne fonctionne pas (facile à tester, à reproduire), plus grand monde ne pourra vous aider s'il y a trop d'âneries à lire, sans pouvoir reproduire sans y passer trop de temps,...

- W