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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
| #!/usr/bin/env python3
# -*- coding: utf-8 -*-
import sqlite3
from dataclasses import dataclass
################################################
# class gestion base
###############################################
class GestionBD(object):
"Mise en place et interfaçage d'une base de données SQLite"
tableInfoKey = ("cid", "name", "type", "notnull", "dflt_value", "pk") #PRAGMA table_info()
index_listKey = ("id?", "name", "unique", "created", "partial") # PRAGMA index_list()
foreignkey_listKey = ("id", "seq", "table", "from", "to", "on_update", "on_delete", "match") # PRAGMA foreign_key_list()
fkeyKey = ('table', 'atribut')
def __init__(self, dbName):
"Établissement de la connexion - Création du curseur"
self.dbName = dbName
def convertSQLtoSQLite(self, req):
"convertion de requete SQL en SQLite"
reqLite = req
reqLite = reqLite.replace(" INT,", " INTEGER,")
reqLite = reqLite.replace(" VARCHAR(50),", " TEXT,")
return reqLite
def executerReq(self, req):
"Exécution de la requête <req>, avec détection d'erreur éventuelle"
connex =sqlite3.connect(self.dbName)
cursor =connex.cursor()
try:
if "SELECT" in req.upper() or "PRAGMA" in req.upper():
cursor.execute(req)
else:
cursor.executescript(req)
ret = cursor.fetchall() # renvoyer une liste de tuples
except Exception as err:
# renvoyer la requête et le message d'erreur système :
msg ="Requête SQL incorrecte :\n{}\nErreur détectée :".format(req)
ret = msg +str(err)
finally:
connex.commit() # On enregistre systématiquement
cursor.close()
connex.close()
return ret
def creaTables(self, dicTables):
"Création des tables de la base de données si elles n'existent pas déjà"
for table in dicTables: # parcours des clés du dictionnaire
req = "CREATE TABLE {} (".format(table)
pk = ""
for descr in dicTables[table]:
nomChamp = descr[0] # libellé du champ à créer
tch = descr[1] # type de champ à créer
if tch == "i":
typeChamp = "INTEGER"
elif tch == "a":
# champ 'clé primaire' (entier incrémenté automatiquement)
typeChamp = "INTEGER PRIMARY KEY AUTOINCREMENT"
pk = nomChamp
elif tch == "k":
# champ 'clé primaire' (entier non automatiquement)
typeChamp = "INTEGER NOT NULL UNIQUE"
pk = nomChamp
elif tch == "r":
typeChamp = "REAL"
else: # pour simplifier, nous considérons
typeChamp = "TEXT" # comme textes tous les autres types
req += "{} {}, ".format(nomChamp, typeChamp)
req = req + """PRIMARY KEY("{}"))""".format(pk)
err = ""
err = self.executerReq(req)
def tableInfo(self):
dtable = {}
dtableid = {}
ltable = self.nameTable()
for i in ltable:
linfo = self.exereq(f"""PRAGMA table_info({i});""")
print(f'linfo: {linfo}')
dtable[i] = {}
dtableid[i] = {}
for t in linfo:
print(f'b linfo t: {t}')
name = t[GestionBD.tableInfoKey.index('name')]
id = t[GestionBD.tableInfoKey.index('cid')]
dtable[i][name] = AtribDef()
dtableid[i][id] = dtable[i][name]
dtable[i][name].type = t[GestionBD.tableInfoKey.index('type')]
dtable[i][name].notnull = t[GestionBD.tableInfoKey.index('notnull')]
dtable[i][name].dflt_value = t[GestionBD.tableInfoKey.index('dflt_value')]
dtable[i][name].pk = t[GestionBD.tableInfoKey.index('pk')]
lfkinfo = self.exereq(f"""PRAGMA foreign_key_list({i});""")
for f in lfkinfo:
print(f"boucle:i={i} f={f}, from={f[GestionBD.foreignkey_listKey.index('from')]} table = {f[GestionBD.foreignkey_listKey.index('table')]}")
dtable[i][f[GestionBD.foreignkey_listKey.index('from')]].fbk = (f[GestionBD.foreignkey_listKey.index('table')], f[GestionBD.foreignkey_listKey.index('to')])
lidinfo = self.exereq(f"""PRAGMA index_list({i});""")
print(f'UNIQUE: {lidinfo}')
for x in range(len(lidinfo)):
name = lidinfo[x][GestionBD.index_listKey.index('name')]
print(f'''b lidinfo id: {x} val: {lidinfo[x][GestionBD.index_listKey.index('unique')]}''')
if lidinfo[x][GestionBD.index_listKey.index('unique')]:
dtable[i][name].unique = True
if lidinfo[x][GestionBD.index_listKey.index('created')] == 'pk':
print(f'''created: {name} {lidinfo[id]}''')
dtableid[i][name].pk = 'fk'
'''
try:
except:
print(f"""
erreur:
lidinfo:{lidinfo}
t: {t}
t[0]: {t[GestionBD.tableInfoKey.index("cid")]}
id:{GestionBD.tableInfoKey.index("cid")}""")
print(f'linfo t: {t} {linfo[t[GestionBD.tableInfoKey.index("cid")]]}')
print(f'dtableunique: {dtable[i][name].unique}')
'''
print(f'dtable: {dtable}')
return dtable
#################################
# Constante géneration base
#################################
dbName = "bd2.sq3" # nom de la base de données
tables ={
"MESURE":(
("DATE","k"),
("VALEUR","i"),
)
}
reqLocMach = """
CREATE TABLE LOCALISATION(
L_KEY INT,
L_ETQ VARCHAR(50) NOT NULL,
L_DOC VARCHAR(50),
L_KEY_1 INT,
PRIMARY KEY(L_KEY),
UNIQUE(L_KEY_1),
UNIQUE(L_ETQ),
FOREIGN KEY(L_KEY_1) REFERENCES LOCALISATION(L_KEY)
);
CREATE TABLE MACHINE(
L_KEY INT,
M_KEY INT,
M_ETQ VARCHAR(50) NOT NULL,
PRIMARY KEY(L_KEY, M_KEY),
UNIQUE(L_KEY),
UNIQUE(M_ETQ),
FOREIGN KEY(L_KEY) REFERENCES LOCALISATION(L_KEY)
);
"""
##################################
# Test
##################################
BD=GestionBD(dbName)
#Creation base
BD=GestionBD('LocMach.sq3')
rep=BD.executerReq('PRAGMA foreign_keys = ON;')
rep=BD.executerReq(reqLocMach)
#app = MyWindow()
#app.mainloop()
lidinfo = BD.executerReq("""PRAGMA index_list("LOCALISATION");""")
print(f'lidinfo: {lidinfo}') |
[sqlite3] Determiner si un attribut est unique dans une table.
Répondre avec citation
Partager