python asynchronous contextmanager

python - Gestor de contexto asíncrono



asynchronous contextmanager (3)

Tengo una API asíncrona que estoy usando para conectar y enviar correo a un servidor SMTP que tiene algo de configuración y se puede desglosar. Por lo tanto, encaja perfectamente en el uso de un contextmanager de contextmanager de la contextmanager de Python 3.

Sin embargo, no sé si es posible escribir porque ambos usan la sintaxis del generador para escribir.

Esto podría demostrar el problema (contiene una combinación de sintaxis de base de rendimiento y espera asíncrona para demostrar la diferencia entre las llamadas asíncronas y los rendimientos al administrador de contexto).

@contextmanager async def smtp_connection(): client = SMTPAsync() ... try: await client.connect(smtp_url, smtp_port) await client.starttls() await client.login(smtp_username, smtp_password) yield client finally: await client.quit()

¿Es este tipo de cosas posibles dentro de python actualmente? y ¿cómo usaría with una declaración as si fuera? Si no es así, ¿hay una forma alternativa en la que pueda lograr esto?


El paquete asyncio_extras tiene una buena solución para esto:

import asyncio_extras @asyncio_extras.async_contextmanager async def smtp_connection(): client = SMTPAsync() ...

Para Python <3.6, también necesitaría el paquete async_generator y reemplazar el yield client con await yield_(client) .


En Python 3.7, podrás escribir:

from contextlib import asynccontextmanager @asynccontextmanager async def smtp_connection(): client = SMTPAsync() ... try: await client.connect(smtp_url, smtp_port) await client.starttls() await client.login(smtp_username, smtp_password) yield client finally: await client.quit()

Hasta que salga la async_generator 3.7, puedes usar el paquete async_generator para esto. En 3.6, puedes escribir:

# This import changed, everything else is the same from async_generator import asynccontextmanager @asynccontextmanager async def smtp_connection(): client = SMTPAsync() ... try: await client.connect(smtp_url, smtp_port) await client.starttls() await client.login(smtp_username, smtp_password) yield client finally: await client.quit()

Y si quieres trabajar todo el camino de regreso a 3.5, puedes escribir:

# This import changed again: from async_generator import asynccontextmanager, async_generator, yield_ @asynccontextmanager @async_generator # <-- added this async def smtp_connection(): client = SMTPAsync() ... try: await client.connect(smtp_url, smtp_port) await client.starttls() await client.login(smtp_username, smtp_password) await yield_(client) # <-- this line changed finally: await client.quit()


Gracias a @jonrsharpe fue capaz de hacer un administrador de contexto asíncrono.

Esto es lo que el mío terminó pareciendo para cualquier persona que quiera un código de ejemplo:

class SMTPConnection(): def __init__(self, url, port, username, password): self.client = SMTPAsync() self.url = url self.port = port self.username = username self.password = password async def __aenter__(self): await self.client.connect(self.url, self.port) await self.client.starttls() await self.client.login(self.username, self.password) return self.client async def __aexit__(self, exc_type, exc, tb): await self.client.quit()

uso:

async with SMTPConnection(url, port, username, password) as client: await client.sendmail(...)

Siéntase libre de señalar si he hecho algo estúpido.