Recherche d'entreprises via l'API recherche-entreprises.api.gouv.fr et insertion dans une base de données Elasticsearch

chercher_entreprises.py

import requests
from pprint import pprint
from elasticsearch import Elasticsearch, helpers


# Connexion à l'instance ElasticSearch
es = Elasticsearch(hosts="http://elastic:changeme@localhost:9200/")

# On crée une fonction pour récupérer les informations des entreprises
# avec comme argument : page pour la page actuelle que l'on récupère
def get_companies(page):

    # L'URL de l'API dans laquelle on utilise l'argument page pour
    # récupérer la page indiquée dans l'appel de la fonction (variable page)
    url = ("https://recherche-entreprises.api.gouv.fr/search"
              "?categorie_entreprise=PME,ETI"
              "&departement=83&page=" + str(page))

    data = requests.get(url).json()

    # On récupère le nombre total de pages pour l'initialisation
    total = data["total_pages"]
    # On récupère les résultats (les détails des entreprises)
    companies = data["results"]

    # On formate notre réponse en dictionnaire avec deux clés :
    # l'une contenant le nb total de pages
    # et l'autre contenant les informations des entreprises
    result = {"total_pages": total, "companies": companies}

    return result


# On initialise notre script, on cherche le nb total de pages
# on ne prete pas attention aux données d'entreprise pour le moment
initial = get_companies(1)
total_pages = initial["total_pages"]

# On initialise la variable dans laquelle on va stocker toutes les entreprises
all_companies = []

# On crée une boucle qui parcourt toutes les pages (voir initialisation)
for i in range(1, total_pages + 1):
    # Pour chaque page, on récupère les données des entreprises
    companies = get_companies(i)["companies"]
    # On les ajoute à notre variable "globale"
    all_companies.extend(companies)

# Insertion d'une liste de documents (ou liste de dictionnaires en Python)
# Et on insère les données d'entreprise dans ES
# dans un index nommé "companies" (index == table en SQL)
result = helpers.bulk(
    es,
    all_companies,
    index="companies",
)