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}') |
Partager