es.davy.ai

Preguntas y respuestas de programación confiables

¿Tienes una pregunta?

Si tienes alguna pregunta, puedes hacerla a continuación o ingresar lo que estás buscando.

Utilizando asyncio para realizar pruebas A/B en Python

02 Aug, 2023 Programación

Supongamos que hay una API que ya está en producción y has creado otra API que quieres probar a través de las solicitudes que se envían a la API de producción. Ahora me pregunto, ¿es posible hacer algo así? (Sé que hay personas que hacen división de tráfico manteniendo dos versiones diferentes de la API para pruebas A/B, etc)

Tan pronto como recibas la solicitud entrante para tu API de producción, haces una solicitud asíncrona a tu nueva API y luego continúas con el resto del código para la API de producción y, justo antes de devolver la respuesta final al llamante, compruebas si tienes los resultados calculados para esa tarea asíncrona que habías creado antes. Si están disponibles, los devuelves en lugar de la API actual.

Me pregunto, ¿cuál es la mejor manera de hacer algo así? ¿Intentamos escribir un decorador para esto o algo más? Me preocupa un poco la cantidad de casos especiales que pueden ocurrir si usamos async aquí. ¿Alguien tiene algún consejo para mejorar el código o el enfoque en general?

¡Gracias por tu tiempo!

Algunos pseudocódigos para el enfoque anterior,

import asyncio

def llamar_api_antigua():
    pass

async def llamar_api_nueva():
    pass

async def principal():
    tarea = asyncio.Task(llamar_api_nueva())

    oldResp = llamar_api_antigua()
    resp = await tarea

    if tarea.done():
        return resp
    else:
        tarea.cancel() # tal vez
        return oldResp

asyncio.run(principal())
Tags:  , , , ,

Answer

  1. Avatar for davy.ai
    davy.ai

    Prácticas recomendadas para probar A/B una API con solicitudes entrantes

    Como experto en IT especializado en Python 3.x, API, programación asíncrona y Python-asyncio, aquí presento mi enfoque sobre cómo realizar pruebas A/B en una API que ya está en funcionamiento y en producción.

    El mejor enfoque sería realizar solicitudes asíncronas a una nueva API y continuar con el código de la API en producción, asegurándose de devolver la respuesta más adecuada al llamador en función de los resultados obtenidos de la tarea asíncrona.

    Una forma de implementar esto es escribir una función decoradora que compruebe el estado de finalización de la tarea asíncrona, devuelva una respuesta alternativa si es necesario e implemente medidas de manejo de errores para asyncio.

    Aquí hay un pseudocódigo que representa el enfoque anterior:

    import asyncio

def decorador(func):
    async def envoltura(*args, **kwargs):
        tarea = asyncio.create_task(func(*args, **kwargs))
        try:
            await asyncio.wait_for(tarea, timeout=1)
            return await tarea
        except asyncio.TimeoutError:
            return None
        except Exception as ex:
            print(f"Se produjo un error: {ex}")
            tarea.cancel()
            raise
    return envoltura

@decorador
async def llamar_nueva_api():
    pass

def llamar_vieja_api():
    pass

async def principal():
    respuesta = await llamar_nueva_api()
    if respuesta is not None:
        return respuesta
    else:
        return llamar_vieja_api()

asyncio.run(principal())

    En el código anterior, el decorador acepta la nueva llamada asíncrona como argumento y la envuelve en una función que maneja el estado de finalización de la tarea asíncrona.

    La función principal luego espera la llamada asíncrona y devuelve la respuesta correspondiente, ya sea de la nueva API (si está disponible) o de la antigua API.

    Recuerde incluir medidas de manejo de errores, como cancelar la tarea si es necesario, para evitar procesos colgados o resultados no deseados.

    En general, este enfoque equilibra la necesidad de probar nuevas API mientras se mantienen las API ya operativas en producción, lo que permite a la organización tomar decisiones basadas en datos sobre el desarrollo futuro de la API.

Comments are closed.