La magie de SQLAlchemy

magie.py

import os
import inspect
from datetime import datetime

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()

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))

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()

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

    sqlite.connect(path)

    engine = create_engine('sqlite:///%s' % path, convert_unicode=True)
    Base.metadata.create_all(bind=engine)
    print(green(u"Base de données créé !"))

# ===========================

import csv
from glob import glob
from datetime import datetime

CSV_PATH = os.path.join("..", "data", "*.csv")
csv_files = glob(CSV_PATH)
csv_files.reverse()

if __name__ == '__main__':
    init_db()
    session = connect(debug=True)

    for csv_file in csv_files:
        # ouverture du fichier
        with open(csv_file) as f:
            csv_content = csv.DictReader(f)

            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()

users.csv