Google API Python Client em aplicações prontas para produção

Como criar softwares melhores com Python e APIs do Google

O Google API Python Client oficial já tem uns bons anos de estrada. Foi escrito para Python 2 e adaptado para rodar no Python 3. Usa o veterano urllib2 e, para fechar, está em modo de manutenção (leia-se: "sem novas funcionalidades").

Novas bibliotecas específicas por produto vêm sendo desenvolvidas, mas ainda cobrem só uma fatia pequena das APIs do Google, com foco principal no GCP. Acho elas muito úteis, mas não cobrem serviços como Google Drive ou Google Directory API. Resultado: por um futuro indefinido, seguimos dependentes do client de baixo nível original que mencionei acima.

Uma das funcionalidades que faltam no Google API Python Client é Thread Safety

Quando seu app fica por um fio (de thread)

Nos últimos anos, venho escrevendo principalmente código Python assíncrono. Projetos como o FastAPI puxam essa abordagem, tanto em tecnologia quanto em mentoria, e facilitam muito escrever serviços web tanto sync quanto async em um único framework. A abordagem assíncrona vicia: ela te livra de ficar pensando no que pode acontecer entre duas linhas de código em outra thread — com o asyncio, é você quem define os pontos de troca de contexto.

Voltando para o mundo síncrono, esqueci da thread-safety por um momento — pelo menos até minha aplicação no Cloud Run começar a quebrar (com direito a Segmentation Fault) e cuspir exceções aleatórias como SSLError, TimeoutError depois de só 25ms, IncompleteRead, etc.

Quando caí na real de que era um problema de thread-safety, comecei a pensar em como resolver. Meu código usa as APIs do G Suite e tenho uma classe de serviço G Suite Manager (kingdom of nouns, eu sei) que inicializo na subida da aplicação e uso ao tratar requisições web no app.

Ficando do lado seguro das threads

A documentação oficial dá algumas dicas de como usar a biblioteca de forma thread-safe — basicamente, gerenciar manualmente os objetos de transporte httplib2 ao chamar APIs do Google e garantir que cada thread use um objeto exclusivo.

Só que não fica claro como aplicar essa dica no contexto de um servidor web, em que cada requisição é tratada por uma thread dedicada que pode ou não continuar viva depois que a requisição termina. Criar um novo objeto httplib2.Http() a cada requisição é desperdício, porque estaríamos abrindo uma nova conexão TCP com o backend da API do Google em toda requisição, além de perder a capacidade do httplib2 de aproveitar ETags para reduzir o payload de resposta dos servidores Google.

No fim, tive a ideia de criar um pool thread-safe próprio de objetos httplib2.Http() e fazer todas as requisições do Google API Python client passarem por ele — chamei de APIConnector.

Olha como ficou:

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

Pouquíssimo código para garantir thread safety, né? Vale notar que nosso gerenciador de pool é thread-safe sem usar nenhum lock — operações atômicas em Python são thread-safe, já que deixamos o GIL cuidar disso pra gente.

E é assim que se usa:

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

Fazer todos os métodos relevantes da API passarem por Managers, como no exemplo acima, pode parecer muito boilerplate, mas, na prática, esses managers acabam fazendo muito mais do que apenas devolver as respostas da API ipsis litteris. No mínimo, eles devem modelar as respostas da API do Google em objetos Python descritivos e adequados (por exemplo, User, Group, etc.), em vez de retornar dicionários puros, que rapidinho deixam o código upstream incompreensível.

Indo além

Se você está curtindo o conteúdo até aqui, vamos para os bônus.

Timeouts

O Google API Python client não oferece uma forma nativa de definir timeouts, mas, como nosso APIConnector já mexe nesses detalhes internos, aproveitamos a deixa para controlar timeouts — como o leitor atento já deve ter notado.

Limpezas

Usar o APIConnector acima "como está" provavelmente vai deixar você com avisos parecidos com o de baixo quando o programa terminar.

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

Isso acontece porque sobram objetos httplib2.Http() não fechados no pool. Dá para resolver adicionando os métodos abaixo ao APIConnector:

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

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

Pronto: os avisos somem e o shutdown fica limpo.

Você também pode reparar que o pool nunca encolhe, o que é verdade. Mas, como meu caso de uso é Cloud Run, onde as instâncias têm vida curta de qualquer jeito, considero um bom trade-off em prol da simplicidade.

Cache

O httplib2 suporta cache para reaproveitar ETags armazenados, usando cabeçalhos If-None-Match ao recuperar um recurso de novo. Isso evita baixar os dados outra vez dos servidores do Google, embora ainda gere um round trip de rede.

O cache do httplib2 é baseado em arquivos e também não é thread-safe. Mesmo assim, queremos ter um cache compartilhado entre os diferentes objetos Http(), já que eles funcionam como um connection pool conversando com a mesma API upstream.

Então, de novo, aproveitando que operações atômicas em objetos Python são thread-safe, dá para montar um cache em memória sem locks para o httplib2 sem dificuldade:

@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

(A interface da classe foi copiada do objeto httplib2.FileCache.)

Agora atualizamos as classes para usar o 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

Por fim, você deve estar se perguntando por que passei cache_discover=False ao montar o service no GSuiteUserManager — a resposta é que essa funcionalidade é bem furada e gera ruído de traceback, como descrito em detalhes aqui e aqui.

Espero que este artigo te ajude a construir softwares melhores com Python e APIs do Google, mesmo a biblioteca oficial tendo algumas arestas para aparar.

O código completo do APIConnector e do MemCache está disponível aqui.