python - Gestor de contexto asíncrono

asynchronous contextmanager

Question

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?

Answer 1

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

Answer 2

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

Answer 3

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.