Google API Python Client para apps de nivel producción

Cómo crear mejor software con Python y las APIs de Google

El Google API Python Client oficial ya tiene sus años. Se escribió para Python 2 y luego se adaptó a Python 3. Usa el viejo conocido urllib2 y, para colmo, hoy se considera en modo de mantenimiento (léase: "sin nuevas funcionalidades").

Se están desarrollando nuevas librerías específicas por producto, pero todavía cubren un rango pequeño de las APIs de Google y se enfocan principalmente en GCP. Me parecen muy útiles, pero no cubren servicios como Google Drive o Google Directory API. Como resultado, por un futuro indefinido vamos a seguir dependiendo del cliente original de bajo nivel que mencioné antes.

Una de las funcionalidades que le falta a Google API Python Client es Thread Safety

Cuando tu app pende de un hilo

En los últimos años he escrito mayormente código async en Python. Proyectos como FastAPI abanderan este enfoque, tanto en lo tecnológico como en lo pedagógico, y permiten escribir servicios web tanto sync como async con un solo framework. El enfoque async es bastante adictivo: te libera de pensar en lo que puede pasar entre dos líneas de código en otro hilo, porque con asyncio eres tú quien define los puntos de cambio de contexto.

Al volver al código sync, simplemente se me olvidó por un momento el tema del thread-safety, al menos hasta que mi aplicación en Cloud Run empezó a caerse (con Segmentation Fault) y a lanzar excepciones aleatorias como SSLError, TimeoutError tras apenas 25 ms, IncompleteRead, etc.

Cuando me di cuenta de que era un problema de thread-safety, me puse a darle vueltas a cómo solucionarlo. Mi código usa las APIs de G Suite y tengo una clase de servicio G Suite Manager (kingdom of nouns, lo sé) que inicializo al arrancar y uso para procesar las solicitudes web de mi app.

Del lado seguro de los hilos

La documentación oficial da algunas recomendaciones para usar la librería de forma thread-safe: gestionar manualmente los objetos de transporte de httplib2 al llamar a las APIs de Google y asegurarse de que cada hilo use uno propio.

Sin embargo, no queda claro cómo aplicar esa recomendación en un servidor web donde cada solicitud se procesa en un hilo dedicado que puede o no seguir vivo después de terminar. Crear un nuevo objeto httplib2.Http() en cada solicitud es un derroche, porque estaríamos abriendo una conexión TCP nueva al backend de Google API en cada solicitud que atendemos y, además, perderíamos la capacidad de httplib2 de aprovechar los ETags para reducir el payload de respuesta de los servidores de Google.

Al final se me ocurrió crear mi propio pool thread-safe de objetos httplib2.Http() y hacer pasar todas las solicitudes del Google API Python Client por ese pool. Lo llamo APIConnector.

Así se ve:

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()

Bastante poco código a cambio de thread safety, ¿no? Vale la pena destacar que nuestro gestor de pool es thread-safe sin usar ningún lock: las operaciones atómicas en Python son thread-safe, porque dejamos que el GIL nos cubra las espaldas.

Así se usa lo anterior:

@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:
        ...

Pasar todos los métodos relevantes de la API por Managers, como en el ejemplo anterior, puede parecer mucho boilerplate, pero en la práctica esos managers harán bastante más que devolver respuestas de la API al pie de la letra. Como mínimo, deberían modelar las respuestas de las APIs de Google en objetos Python descriptivos (por ejemplo, User, Group, etc.) en lugar de devolver simples diccionarios, que vuelven el código de más arriba ilegible muy rápido.

Cómo mejorarlo aún más

Si hasta aquí te resulta interesante, vamos con las secciones de contenido extra.

Timeouts

Google API Python Client no ofrece una forma nativa de definir timeouts, pero como nuestro APIConnector ya se mete en sus entrañas, aprovechamos para controlarlos, como ya habrá notado el lector atento.

Limpieza

Usar el APIConnector anterior "tal cual" probablemente te deje con advertencias como la de abajo cuando termine tu programa.

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)>

Esto se debe a que en el pool quedan objetos httplib2.Http() sin cerrar. Lo solucionamos agregando los siguientes métodos a nuestro APIConnector:

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

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

Ya no aparecen las advertencias y el cierre queda limpio.

También puede que notes que el pool nunca se reduce, lo cual es cierto, pero como mi caso de uso es Cloud Run, donde las instancias son de vida corta de todos modos, me parece un buen trade-off a favor de la simplicidad.

Caché

Httplib2 soporta caché para usar ETags almacenados con cabeceras If-None-Match al volver a pedir un recurso. Esto evita tener que descargar los datos otra vez desde los servidores de Google, aunque igual implica un round trip de red.

La caché de httplib2 está basada en archivos y tampoco es thread-safe. Aun así, sí queremos una caché compartida entre los distintos objetos Http(), ya que los usamos como un pool de conexiones para hablar con la misma API upstream.

Así que, una vez más, aprovechando que las operaciones atómicas sobre objetos Python son thread-safe, podemos armar fácilmente una caché en memoria sin locks para 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

(La interfaz de la clase se copió del objeto httplib2.FileCache)

Y ahora actualizamos nuestras clases para usar la caché:

@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

Por último, quizá te preguntes por qué pasé cache_discover=False al construir el servicio en GSuiteUserManager. La respuesta es que esa funcionalidad está bastante rota y genera ruido en los tracebacks, como se describe en detalle aquí y aquí.

Espero que este artículo te ayude a crear mejor software con Python y las APIs de Google, a pesar de que la librería oficial tenga un par de aristas.

El código completo de APIConnector y MemCache está disponible aquí.