Purge_Registres_D-c-s_INSEE/purge-registres-deces-insee/bdd_insee.py

"""Classe d'accès aux données INSEE dans la base SQLite

"""

import os
import peewee


# Objet de connexion à la base
database = peewee.SqliteDatabase(None)


class BddInsee:
    """Classe encapsulant les accès aux données.

    """

    def __init__(self, chemin_base_donnees):
        """Initialisation

        :param chemin_base_donnees: chemin vers le fichier SQLite

        """
        new = not os.path.isfile(chemin_base_donnees)        
        database.init(chemin_base_donnees)
        if new:
            database.create_tables([self.Person])

    class Person(peewee.Model):
        """Classe représentant une personne dans la base

        """
        last_name = peewee.CharField()
        first_name = peewee.CharField()
        is_woman = peewee.BooleanField()
        annee_naissance = peewee.IntegerField()
        mois_naissance = peewee.IntegerField()
        jour_naissance = peewee.IntegerField()
        code_lieu_naissance = peewee.CharField()
        commune_naissance = peewee.CharField()
        pays_naissance = peewee.CharField()
        annee_deces = peewee.IntegerField()
        mois_deces = peewee.IntegerField()
        jour_deces = peewee.IntegerField()
        code_lieu_deces = peewee.CharField()
        numero_act_deces = peewee.CharField()

        class Meta:
            database = database

    def find_person(self, first_name, last_name, maiden_name, annee_naissance, mois_naissance, jour_naissance):
        """Rechercher une personne dans la base

        :param first_name: prénom de la personne à rechercher
        :param last_name: nom de famille de la personne à rechercher
        :param maiden_name: nom de jeune fille de la personne à rechercher (None si non défini)
        :param annee_naissance: année de naissance de la personne à rechercher
        :param mois_naissance: mois de naissance de la personne à rechercher
        :param jour_naissance: jour de naissance de la personne à rechercher
        :returns: liste contenant un dictionnaire par personne trouvée

        """
        if maiden_name:
            return self.Person.select().where((self.Person.annee_naissance == int(annee_naissance))
                                         & (self.Person.mois_naissance == int(mois_naissance))
                                         & (self.Person.jour_naissance == int(jour_naissance))
                                         & (self.Person.first_name.contains(first_name.upper()))
                                         & ((self.Person.last_name.contains(last_name.upper()))
                                            | self.Person.last_name.contains(maiden_name.upper())))
        else:
            return self.Person.select().where((self.Person.annee_naissance == int(annee_naissance))
                                         & (self.Person.mois_naissance == int(mois_naissance))
                                         & (self.Person.jour_naissance == int(jour_naissance))
                                         & (self.Person.first_name.contains(first_name.upper()))
                                         & (self.Person.last_name.contains(last_name.upper())))

    def import_data_list(self, data_list):
        """Insérer des données dans la base

        :param data_list: liste de dictionnaires contenant les
        informations à insérer

        """
        with database.atomic():
            for batch in peewee.chunked(data_list, 70):
                self.Person.insert_many(batch).execute()