Google API Python Client en production : le guide pratique

Comment concevoir de meilleurs logiciels avec Python et les API Google

Le Google API Python Client officiel commence à dater. Écrit pour Python 2 puis adapté à Python 3, il s'appuie sur le vénérable urllib2 et se trouve aujourd'hui en mode maintenance (comprenez : plus de nouvelles fonctionnalités).

De nouvelles bibliothèques dédiées à chaque produit voient le jour, mais elles ne couvrent encore qu'une petite partie des API Google, principalement autour de GCP. Je les trouve très utiles, mais elles laissent de côté des services comme Google Drive ou Google Directory API. Résultat : nous restons largement dépendants du client bas niveau d'origine évoqué plus haut, et ce pour un avenir indéterminé.

L'une des fonctionnalités qui manquent au Google API Python Client, c'est la thread safety

Quand votre application ne tient qu'à un fil (d'exécution)

Ces dernières années, j'ai surtout écrit du code Python asynchrone. Des projets comme FastAPI sont à la pointe de cette approche, à la fois sur le plan technique et pédagogique, et permettent d'écrire aussi bien des services web sync qu'async avec un seul framework. L'approche async devient vite addictive : elle vous évite d'avoir à anticiper ce qui peut survenir entre deux lignes de code dans un autre thread — avec asyncio, c'est vous qui définissez les points de bascule de contexte.

En revenant au code synchrone, j'ai tout simplement oublié la thread safety pendant un moment, du moins jusqu'à ce que mon application Cloud Run se mette à planter (genre Segmentation Fault) et à lever des exceptions aléatoires : SSLError, TimeoutError au bout de seulement 25 ms, IncompleteRead, etc.

Une fois le problème de thread safety identifié, restait à le résoudre. Mon code utilise les API G Suite et j'ai une classe de service G Suite Manager (le royaume des noms, je sais) que j'initialise au démarrage et que j'utilise pendant le traitement des requêtes web.

Rester du bon côté de la thread safety

La documentation officielle donne quelques conseils pour utiliser la bibliothèque de manière thread-safe : gérer manuellement les objets de transport httplib2 lors des appels aux API Google, et veiller à ce que chaque thread dispose du sien.

Reste à savoir comment appliquer ce conseil dans le contexte d'un serveur web où chaque requête est traitée dans un thread dédié, qui peut survivre ou non au traitement. Créer un nouvel objet httplib2.Http() à chaque requête est un gaspillage : cela revient à ouvrir une nouvelle connexion TCP vers le backend des API Google à chaque requête traitée, et à perdre la capacité de httplib2 à exploiter les ETags pour réduire la taille des réponses des serveurs Google.

J'ai fini par imaginer mon propre pool thread-safe d'objets httplib2.Http() et faire transiter par ce pool toutes les requêtes du client Python des API Google — je l'ai baptisé APIConnector.

Voici à quoi il ressemble :

from google_auth_httplib2 import AuthorizedHttp
from googleapiclient.http import HttpRequest
import google.auth
import httplib2

@dataclass
class APIConnector:
    factory: Callable[[], AuthorizedHttp]
    pool: List[AuthorizedHttp] = field(default_factory=[])

    @classmethod
    def new(
        cls,
        credentials: google.auth.Credentials,
        initial_size: int = 5,
        timeout_seconds: int = 10,
    ) -> APIConnector:

        factory = lambda: AuthorizedHttp(
          credentials,
          http=httplib2.Http(timeout=timeout_seconds)
        )

        pool: List[AuthorizedHttp] = []
        for i in range(initial_size):
            pool.append(factory())

        return cls(factory, pool=pool)

    def execute(self, req: HttpRequest) -> Any:
        http: Optional[AuthorizedHttp] = None
        try:
            http = self._provision_http()
            return req.execute(http=http)
        finally:
            if http:
                self.pool.append(http)

    def _provision_http(self) -> AuthorizedHttp:
        try:
            return self.pool.pop()
        except IndexError:
            logger.info("Pool exhausted. Creating new transport")
            return self.factory()

Peu de code pour gagner la thread safety, n'est-ce pas ? À noter : notre gestionnaire de pool est thread-safe sans aucun verrou — les opérations atomiques en Python sont thread-safe, puisque c'est le GIL qui s'en charge à notre place.

Voici comment l'utiliser :

@dataclass
class GSuiteUserManager:
    api: APIConnector
    users: googleapiclient.discovery.Resource
    domain: str

    @classmethod
    def new(cls, domain, credentials) -> GSuiteUsersManager:
        api = APIConnector.new(Credentials)
        service = googleapiclient.discovery.build(
            "admin",
            "directory_v1",
            credentials=credentials,
            cache_discovery=False,
        )
        users = service.users()
        return cls(api=api, users=users, domain=domain)

    def list(self) -> dict:
        return self.api.execute(
            self.users.list(domain=self.domain)
        )

    def get(self, email: str) -> dict:
        ...

Faire transiter toutes les méthodes d'API pertinentes par des Managers, comme dans l'exemple ci-dessus, peut sembler très répétitif, mais en pratique ces managers font bien plus que renvoyer les réponses de l'API telles quelles. Au minimum, ils devraient modéliser les réponses des API Google sous forme d'objets Python descriptifs (par ex. User, Group, etc.) plutôt que de retourner de simples dictionnaires, qui rendent rapidement le code en aval illisible.

Aller encore plus loin

Si le sujet vous intéresse, passons aux sections bonus.

Timeouts

Le client Python des API Google n'offre aucun moyen natif de définir des timeouts, mais comme notre APIConnector plonge de toute façon dans ses entrailles, autant en profiter pour les gérer — un lecteur attentif l'aura déjà remarqué.

Nettoyage

Utiliser l'APIConnector ci-dessus tel quel laissera probablement apparaître des avertissements comme celui-ci à la fin du programme.

sys:1: ResourceWarning: unclosed <ssl.SSLSocket fd=5, family=AddressFamily.AF_INET, type=SocketKind.SOCK_STREAM, proto=6, laddr=('192.168.1.1', 54988), raddr=('142.250.66.173', 443)>

C'est parce que des objets httplib2.Http() n'ont pas été fermés dans notre pool. Corrigeons cela en ajoutant les méthodes suivantes à notre APIConnector :

def close(self) -> None:
    for ahttp in self.pool:
        ahttp.http.close()

def __del__(self) -> None:
    self.close()

Les avertissements disparaissent et l'arrêt est propre.

Vous remarquerez aussi que le pool ne se réduit jamais — c'est exact, mais comme mon cas d'usage est Cloud Run, où les instances sont de toute façon éphémères, le compromis me semble acceptable au profit de la simplicité.

Cache

Httplib2 prend en charge la mise en cache afin d'exploiter les ETags stockés via les en-têtes If-None-Match lors de la récupération d'une ressource déjà connue. Cela évite de retélécharger les données depuis les serveurs Google, même si un aller-retour réseau reste nécessaire.

Le cache de httplib2 repose sur des fichiers et n'est pas non plus thread-safe. Or, nous voulons un cache partagé entre les différents objets Http(), puisque nous les utilisons comme un pool de connexions vers la même API en amont.

Là encore, en exploitant le fait que les opérations atomiques sur les objets Python sont thread-safe, on peut facilement construire un cache mémoire sans verrou pour httplib2 :

@dataclass

class MemCache:
    data: dict[Hashable, Any] = field(default_factory=dict)

    def get(self, key: Hashable) -> Any:
        if hit := self.data.get(key, None):
            logger.debug("Cache hit", key=key)
        return hit

    def set(self, key: Hashable, data: Any) -> None:
        self.data[key] = data

    def delete(self, key):
        try:
            del self.data[1]
        except KeyError:
            pass

(L'interface de la classe est reprise de l'objet httplib2.FileCache.)

Mettons maintenant nos classes à jour pour utiliser le cache :

@dataclass
class APIConnector:
    ...

    @classmethod
    def new(
        ...
        cache: Optional[MemCache] = None,
    ) -> APIConnector:

        factory = lambda: AuthorizedHttp(
            credentials,
            http=httplib2.Http(
                timeout=timeout_seconds,
                cache=cache,
            ),
        )
        ...

class GSuiteUserManager:
    ...

    @classmethod
    def new(
        cls, domain, credentials, use_cache: bool = True
    ) -> GSuiteUsersManager:
        cache = MemCache() if use_cache else None
        api = APIConnector.new(credentials, cache=cache)
        ...

Discovery Cache

Enfin, vous vous demandez peut-être pourquoi j'ai passé cache_discover=False lors de la construction du service dans GSuiteUserManager : tout simplement parce que cette fonctionnalité est franchement défaillante et pollue les tracebacks, comme décrit en détail ici et là.

J'espère que cet article vous aidera à concevoir de meilleurs logiciels avec Python et les API Google, malgré les quelques aspérités de la bibliothèque officielle.

Le code complet d'APIConnector et de Memcache est disponible ici.