Introduction à Python 3 et au framework web Flask par l'exemple

# suppression de la table [personnes]
drop table if exists personnes
# création de la table personnes
create table personnes (id int primary key, prenom varchar(30) not null, nom varchar(30) not null, age integer not null, unique (nom,prenom))
# insertion de deux personnes
insert into personnes(id, prenom, nom, age) values(1, 'Paul','Langevin',48)
insert into personnes(id, prenom, nom, age) values (2, 'Sylvie','Lefur',70)
# affichage de la table
select prenom, nom, age from personnes
# erreur volontaire
xx
# insertion de trois personnes
insert into personnes(id, prenom, nom, age) values (3, 'Pierre','Nicazou',35)
insert into personnes(id, prenom, nom, age) values (4, 'Geraldine','Colou',26)
insert into personnes(id, prenom, nom, age) values (5, 'Paulette','Girond',56)
# affichage de la table
select prenom, nom, age from personnes
# liste des personnes par ordre alphabétique des noms et à nom égal par ordre alphabétique des prénoms
select nom,prenom from personnes order by nom asc, prenom desc
# liste des personnes ayant un âge dans l'intervalle [20,40] par ordre décroissant de l'âge
# puis à âge égal par ordre alphabétique des noms et à nom égal par ordre alphabétique des prénoms
select nom,prenom,age from personnes where age between 20 and 40 order by age desc, nom asc, prenom asc
# insertion de mme Bruneau
insert into personnes(id, prenom, nom, age) values(6, 'Josette','Bruneau',46)
# mise à jour de son âge
update personnes set age=47 where nom='Bruneau'
# liste des personnes ayant Bruneau pour nom
select nom,prenom,age from personnes where nom='Bruneau'
# suppression de Mme Bruneau
delete from personnes where nom='Bruneau'
# liste des personnes ayant Bruneau pour nom
select nom,prenom,age from personnes where nom='Bruneau'

def configure():
    import os

    # chemin absolu du dossier de ce script
    script_dir = os.path.dirname(os.path.abspath(__file__))

    # configuration des dossiers du syspath
    absolute_dependencies = [
        # dossiers locaux
        f"{script_dir}/shared",
    ]

    # fixation du syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # configuration de l'application
    config = {
        # sgbd gérés
        "sgbds": {
            "mysql": {
                # connecteur du sgbd
                "sgbd_connector": "mysql.connector",
                # nom du module rassemblant des fonctions de gestion du sgbd
                "sgbd_fonctions": "any_module",
                # identifiants de la connexion
                "user": "admpersonnes",
                "password": "nobody",
                "host": "localhost",
                "database": "dbpersonnes"
            },
            "postgresql": {
                # connecteur du sgbd
                "sgbd_connector": "psycopg2",
                # nom du module rassemblant des fonctions de gestion du sgbd
                "sgbd_fonctions": "any_module",
                # identifiants de la connexion
                "user": "admpersonnes",
                "password": "nobody",
                "host": "localhost",
                "database": "dbpersonnes"
            }
        },
        # fichier de commandes SQL
        "commands_filename": f"{script_dir}/data/commandes.sql"
    }
    # syspath de l'application
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    # on rend la config
    return config

# on configure l'application
import config

config = config.configure()

# le syspath est configuré - on peut faire les imports
import importlib
import sys

# vérification de la syntaxe de l'appel
# argv[0] sgbd_name true / false
# il faut 3 paramètres
args = sys.argv
erreur = len(args) != 3
if not erreur:
    # on récupère les deux paramètres qui nous intéressent
    sgbd_name = args[1].lower()
    with_transaction = args[2].lower()
    # vérification des deux paramètres
    erreur = (with_transaction != "true" and with_transaction != "false") \
             or sgbd_name not in config["sgbds"].keys()
# erreur ?
if erreur:
    print(f"syntaxe : {args[0]} (1) sgbd_name (2) true / false")
    sys.exit()

# configuration du sgbd_name
sgbd_config = config["sgbds"][sgbd_name]
# connecteur du sgbd_name
sgbd_connector = importlib.import_module(sgbd_config["sgbd_connector"])
# bibliothèque de fonctions
lib = importlib.import_module(sgbd_config["sgbd_fonctions"])


# calcul d'un texte à afficher
with_transaction = with_transaction == "true"
if with_transaction:
    texte = "avec transaction"
else:
    texte = "sans transaction"

# affichage
commands_filename=config['commands_filename']
print("--------------------------------------------------------------------")
print(f"Exécution du fichier SQL {commands_filename} {texte}")
print("--------------------------------------------------------------------")

# exécution des ordres SQL du fichier
connexion = None
try:
    # connexion
    connexion = sgbd_connector.connect(
        host=sgbd_config['host'],
        user=sgbd_config['user'],
        password=sgbd_config['password'],
        database=sgbd_config['database'])
    # exécution du fichier des commandes SQL
    erreurs = lib.execute_file_of_commands(sgbd_connector, connexion, commands_filename, suivi=True, arrêt=False,
                                           with_transaction=with_transaction)
except (sgbd_connector.InterfaceError, sgbd_connector.DatabaseError) as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    # fermeture de la connexion
    if connexion:
        connexion.close()

# affichage nombre d'erreurs
print("--------------------------------------------------------------------")
print(f"Exécution terminée")
print("--------------------------------------------------------------------")
print(f"Il y a eu {len(erreurs)} erreur(s)")
# affichage des erreurs
for erreur in erreurs:
    print(erreur)

# ---------------------------------------------------------------------------------

def afficher_infos(curseur):
    …


# ---------------------------------------------------------------------------------
def execute_list_of_commands(sgbd_connector, connexion, sql_commands: list,
                             suivi: bool = False, arrêt: bool = True, with_transaction: bool = True):
    …

    # initialisations
    curseur = None
    connexion.autocommit = not with_transaction
    erreurs = []
    try:
        # on demande un curseur
        curseur = connexion.cursor()
        # exécution des sql_commands SQL contenues dans sql_commands
        # on les exécute une à une
        for command in sql_commands:
            # on élimine les blancs de début et de fin de la commande courante
            command = command.strip()
            # a-t-on une commande vide ou un commentaire ? Si oui, on passe à la commande suivante
            if command == '' or command[0] == "#":
                continue
            # exécution de la commande courante
            error = None
            try:
                curseur.execute(command)
            except (sgbd_connector.InterfaceError, sgbd_connector.DatabaseError) as erreur:
                error = erreur
            # y-a-t-il eu une erreur ?
            …


# ---------------------------------------------------------------------------------
def execute_file_of_commands(sgbd_connector, connexion, sql_filename: str,
                             suivi: bool = False, arrêt: bool = True, with_transaction: bool = True):
    …

    # exploitation du fichier SQL
    try:
        # ouverture du fichier en lecture
        file = open(sql_filename, "r")
        # exploitation
        return execute_list_of_commands(sgbd_connector, connexion, file.readlines(), suivi, arrêt, with_transaction)
    except BaseException as erreur:
        # on rend un tableau d'erreurs
        return [f"Le fichier {sql_filename} n'a pu être être exploité : {erreur}"]
    finally:
        pass