axione-elig-test/address_finder/api.py

import sqlite3
import sys
import json
from .model import Commune,FantoirVoie

# DB with addresses info
DB_ADDRESSES_PATH_ENV="DB_ADDRESSES_PATH"
DB_ADDRESSES_DEFAULT_PATH="/etc/fantoir.sqlite"

# Table for insee codes
DB_TABLE_INSEE_NAME="insee"
DB_COL_COMMUNE_INSEE="Code_commune_INSEE"
DB_COL_COMMUNE_NAME="Nom_commune"
DB_COL_COMMUNE_POSTE="Code_postal"

# Table for Fantoir voies (code Rivoli)
DB_TABLE_FANTOIR_NAME="keyv"
DB_COL_FANTOIR_INSEE="key"
DB_FANTOIR_INSEE_KEY_SUFFIX="keyv:"

# Utility to find an address
class AddressFinder:

    def __init__(self, db_addresses_sqlite_path: str):
        self.dbPath = db_addresses_sqlite_path
        print("DB addresses Path : " + self.dbPath)

    def getCommunesFromNameOrZip(self, communeNameOrZip: str) -> list[Commune]:
        con = sqlite3.connect(self.dbPath)
        con.row_factory = sqlite3.Row
        cur = con.cursor()
        communes: list[Commune] = []
        try:
          if communeNameOrZip is None:
            cur.execute(f"SELECT * from \"{DB_TABLE_INSEE_NAME}\"")
          else:
            cur.execute(f"SELECT * from \"{DB_TABLE_INSEE_NAME}\" WHERE {DB_COL_COMMUNE_NAME}=\"{communeNameOrZip}\" COLLATE nocase OR {DB_COL_COMMUNE_POSTE}=\"{communeNameOrZip}\"")
        except sqlite3.OperationalError as err:
          print("Error querying DB : {0}".format(err), file=sys.stderr)
          return []
        rows = [dict(row) for row in cur.fetchall()]
        con.close()
        for row in rows:
          commune=Commune(
            codeInsee=row[DB_COL_COMMUNE_INSEE],
            nom=row[DB_COL_COMMUNE_NAME],
            codeZip=row[DB_COL_COMMUNE_POSTE])
          communes.append(commune)
        return communes

    def getCommuneFantoirVoies(self, communeInseeCode: str) -> list[FantoirVoie]:
        
        # Extract data from DB
        con = sqlite3.connect(self.dbPath)
        con.row_factory = sqlite3.Row
        cur = con.cursor()
        try:
          cur.execute(f"SELECT value from \"{DB_TABLE_FANTOIR_NAME}\" WHERE {DB_COL_FANTOIR_INSEE}=\"{DB_FANTOIR_INSEE_KEY_SUFFIX}{communeInseeCode}\"")
        except sqlite3.OperationalError as err:
          print("Error querying DB : {0}".format(err), file=sys.stderr)
          return []
        data_raw = cur.fetchone()
        con.close()

        ## Get JSON payload
        
        fantoir_dict = []
        # Check if data where found
        if data_raw is not None:
          data = dict(data_raw)
          # Extract the data behind "value" which is a JSON structure
          data_dict=json.loads(data.get("value"))
          # In extracted JSON data, the interesting payload is behind "value" key
          fantoir_dict = data_dict.get("value")
        else:
          print("Did not found any data matching Insee code " + str(communeInseeCode))
        
        # Return the json dump
        return fantoir_dict