Québec-Python


La magie de SQLAlchemy

Par Bernard Chhun le dim 09 février 2014

Voici donc une petite recette bien savoureuse pour extraire des données d'un fichier CSV et l'insérer dans une base de données.

3 modules seront utilisés:

  • SQLAlchemy: L'ORM
  • Fabric: Pas vraiment nécessaire dans ce cas-ci, mais je l'utilise pour afficher des messages de couleur dans le terminal.
  • PySQLite: Pour la connexion à une base de données SQLite

Roulez les commandes suivantes pour les installer dans votre environnement virtuel:

$ virtualenv venv
$ source venv/bin/activate
$ pip install sqlalchemy fabric pysqlite

Nous avons donc les données suivantes:

country,nas,fullname,created_at,birthday,email
Maldives,122,Tre Kunde,Sat Oct 04 1980 19:18:34,23/5/2007,Diamond@buster.co.uk
Christmas Island,123,Nina Sauer I,Sun Sep 11 2011 07:11:11,6/3/2008,Fae@haskell.biz
Western Sahara,124,Jerod O'Reilly,Mon May 24 1993 03:34:03,27/5/1982,Eloy@jailyn.tv
Maldives,125,Aliza Nienow,Fri Nov 29 1996 07:39:35,26/8/1993,Johnson_Marvin@lucius.io
Sierra Leone,126,Litzy Altenwerth MD,Sun Jun 06 1993 04:05:20,14/1/1981,Nicolas@adolfo.com
Peru,127,Willow Bahringer,Tue Sep 15 1992 12:20:10,4/12/1988,Jesse@christy.org
Andorra,128,Ward Bode,Mon Oct 12 1987 17:56:50,10/8/1992,Brandon@helen.co.uk
Philippines,129,Maureen Boehm,Wed Sep 09 1998 00:22:14,7/6/1996,Deondre.Friesen@elroy.ca
United Arab Emirates,130,Linda Bins,Sat May 09 1992 06:41:04,15/1/1984,Jessy@cassidy.me

Du code de base pour SQLAlchemy:

import os
import inspect

from fabric.colors import red, green

from sqlite3 import dbapi2 as sqlite

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import (
    Column, Integer, String, DateTime, Date
)

CURRENT_DIRECTORY = os.path.dirname(
    inspect.getfile(inspect.currentframe())
)

DATABASE_NAME = "users"

Base = declarative_base()

Une classe qui fera le lien entre notre code et la base de données SQLite:

class User(Base):
    __tablename__ = "users"
    id = Column("id", Integer, primary_key = True)
    nas = Column("nas", Integer)
    country = Column("country", String(200))
    fullname = Column("fullname", String(200))
    created_at = Column("created_at", DateTime)
    birthday = Column("birthday", Date)
    email = Column("email", String(200))

Une fonction de création de bd:

def init_db():
    path = os.path.join(CURRENT_DIRECTORY, "{}.db".format(DATABASE_NAME))

    if os.path.isfile(path):
        print(red(u"La base de données existe déjà..."))
        print(red(u"Sa création est annulé"))
        return

    # Création d'un fichier sqlite s'il n'existe pas déjà
    sqlite.connect(path)

    engine = create_engine('sqlite:///%s' % path, convert_unicode=True)
    # magie ! création des tables selon les models que vous aurez déclaré
    # précédemment
    Base.metadata.create_all(bind=engine)

    print(green(u"Base de données créé !"))

Et finalement, une fonction de création de session:

def connect(debug=False):
    path = os.path.join(CURRENT_DIRECTORY, "{}.db".format(DATABASE_NAME))
    engine = create_engine(
        'sqlite+pysqlite:///{}'.format(path),
        module=sqlite,
        echo=debug
    )
    SessionMakerInstance = sessionmaker(bind=engine)
    return SessionMakerInstance()

Maintenant que nous avons tous les ingrédients, voici la recette elle-même. Prenez le temps de lire les commentaires si vous êtes un débutant:

import csv
from glob import glob
from datetime import datetime

# obtention de tous les fichiers *.csv dans le dossier courant
CSV_PATH = os.path.join("*.csv")
csv_files = glob(CSV_PATH)
csv_files.reverse()

# création de la base de données
init_db()

# Création de la connexion à la base de données
session = connect(debug=True)

# Pour chacun des fichiers csv dans le dossier data
for csv_file in csv_files:

    # ouverture du fichier. à noté: fermeture implicite à la sortie du "with"
    with open(csv_file) as f:
        # utilisation de DictReader pour transformer chaque ligne
        # du fichier texte en dictionnaire
        # ex:
        # Cette ligne du fichier csv:
        # Maldives,122,Tre Kunde,Sat Oct 04 1980 19:18:34,23/5/2007,Diamond@buster.co.uk
        #
        # donne ceci:
        # {
        #    "country": Maldives,
        #    "nas": 122,
        #    "fullname": "Tre kunde",
        #    "created_at": "Sat Oct 04 1980 19:18:34",
        #    "birthday": "23/5/2007",
        #    "email": "Diamond@buster.co.uk"
        # }
        csv_content = csv.DictReader(f)

        # à l'aide d'un generator, instanciation d'un user pour chaque ligne du DictReader
        session.add_all((
            User(
                nas=row["nas"],
                country=row["country"],
                fullname=row["fullname"],
                # Tue Dec 31 1996 11:35:43
                created_at=datetime.strptime(
                    row["created_at"], "%a %b %d %Y %H:%M:%S"
                ).date(),
                birthday=datetime.strptime(
                    row["birthday"], "%d/%m/%Y"
                ).date(),
                email=row["email"]
            ) for row in csv_content
        ))

session.commit()

Vous pouvez prendre le temps de copier-coller tous ces fragments dans un seul et même fichier ou bien télécharger un fichier zip avec le même contenu... Si vous le téléchargez, vous n'avez qu'à rouler le fichier magie.py par la suite.

N'hésitez pas à nous poser des questions ou a partager votre expérience avec cette recette dans la liste de diffusions ou sur les réseaux sociaux.

L'équipe de Québec-Python