Endpoints to find a Fantoir address

.gitignore

@@ -0,0 +1,2 @@
+__pycache__
+/elig-test.ini

address_finder/api.py

@@ -0,0 +1,71 @@
+import sqlite3
+import os
+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 getCommunesFromNameOrZipJson(self, communeNameOrZip: str) -> list[Commune]:
+        con = sqlite3.connect(self.dbPath)
+        con.row_factory = sqlite3.Row
+        cur = con.cursor()
+        communes: list[Commune] = []
+        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}\"")
+        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 getCommuneFantoirVoiesJson(self, communeInseeCode: str) -> list[FantoirVoie]:
+        
+        # Extract data from DB
+        con = sqlite3.connect(self.dbPath)
+        con.row_factory = sqlite3.Row
+        cur = con.cursor()
+        cur.execute(f"SELECT value from \"{DB_TABLE_FANTOIR_NAME}\" WHERE {DB_COL_FANTOIR_INSEE}=\"{DB_FANTOIR_INSEE_KEY_SUFFIX}{communeInseeCode}\"")
+        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

address_finder/model.py

@@ -0,0 +1,17 @@
+from typing import TypedDict
+
+class Commune(TypedDict):
+  codeInsee: str
+  nom: str
+  codeZip: str
+
+class FantoirVoie(TypedDict):
+  id: str
+  dateAjout: int
+  libelle: list[str]
+  typeVoie:str
+  codeCommune: str
+  codeFantoir: str
+  cleRivoli: str
+  nomCommune: str
+  predecesseur: bool

config/config.py

@@ -2,10 +2,11 @@ import configparser
 
 
 class Config:
-    def __init__(self, username, password, source_addr):
+    def __init__(self, username, password, source_addr, db_addresses_sqlite_path):
         self.username = username
         self.password = password
         self.source_addr = source_addr
+        self.db_addresses_sqlite_path = db_addresses_sqlite_path
         self.debug = False
 
 
@@ -16,4 +17,5 @@ def parse_config(cfgPath):
     username = cfg.get("API", "username")
     passwd = cfg.get("API", "password")
     source_addr = cfg.get("API", "source_addr")
-    return Config(username, passwd, source_addr)
+    db_addresses_sqlite_path = cfg.get("ADDRESSES","db_addresses_sqlite_path")
+    return Config(username, passwd, source_addr,db_addresses_sqlite_path)

elig-test.ini.sample

@@ -3,4 +3,6 @@
   password = xxx
   # Whitelisted IP address from axione
   # to send the requests from.
-  source_addr = xxx.xxx.xxx.xxx
+  source_addr = xxx.xxx.xxx.xxx
+[ADDRESSES]
+  db_addresses_sqlite_path = path/to/db.sqlite

run-dev-server

@@ -1,3 +1,3 @@
 #!/usr/bin/env bash
 
-DEBUG=true CONFIG=./elig-test.ini.sample FLASK_APP=webapp poetry run flask run --reload
+DEBUG=true CONFIG=./elig-test.ini FLASK_APP=webapp poetry run flask run --reload

webapp.py

@@ -1,8 +1,10 @@
 import os
 from flask import Flask, render_template, request, escape
+import json
 
-from axione_api.config import parse_config
+from config.config import parse_config
 from axione_api.api import query_axione_pto, parse_response
+from address_finder.api import AddressFinder
 
 def load_config():
     cfg_path = os.environ.get("CONFIG", "/etc/ftth-elig/conf.ini")
@@ -19,6 +21,9 @@ def load_config():
     return cfg
 
 cfg = load_config()
+
+addressFinder = AddressFinder(cfg.db_addresses_sqlite_path)
+
 app = Flask(__name__)
 
 @app.route("/", methods=['GET'])
@@ -30,3 +35,23 @@ def show_result():
     pto = escape(request.form['pto'])
     result = parse_response(query_axione_pto(cfg, pto))
     return render_template("result.html", pto=pto, result=result)
+
+@app.route("/addresses/communes", methods=['GET'])
+def get_communes():
+    to_search=request.args.get('s')
+    print(to_search)
+    communes=addressFinder.getCommunesFromNameOrZipJson(to_search)
+    response = app.response_class(
+        response=json.dumps(communes),
+        mimetype='application/json'
+    )
+    return response
+
+@app.route("/addresses/fantoirvoies/<codeInsee>", methods=['GET'])
+def get_fantoir_voies(codeInsee):
+    fantoirVoies=addressFinder.getCommuneFantoirVoiesJson(codeInsee)
+    response = app.response_class(
+        response=json.dumps(fantoirVoies),
+        mimetype='application/json'
+    )
+    return response