El poder del tipado en Python, funciones sin código.

En Python existen distintas técnicas para analizar el código, y con cada una de ellas podremos hacer distintas cosas con mayor o menor dificultad. En este post vamos a ver cómo podemos hacer que nuestro código tenga un comportamiento en función a los tipos de las variables que definen una función, pero para llegar a ese punto primero veremos distintas maneras de analizar y modificar nuestro código en tiempo de ejecución.

Por último presentaremos una librería que, haciendo uso de un decorador, nos permite escribir clientes web solo definiendo nuestras funciones, sin nada de código dentro de ellas, gracias al tipado de sus valores.

Proceso de tokenizado

Empezaremos por las librerías del compilador y seguiremos el orden de compilación.
El primer paso sería el análisis léxico del código. Para ello tenemos la librería tokenize.

En este ejemplo vemos cómo modificar una función para cambiar la operación suma por una multiplicación:

import inspect, tokenize
from io import BytesIO

def suma(a: int, b: int) -> int:  # Just a comment
   return a + b

code: str = inspect.getsource(suma)
tokens = list(tokenize.tokenize(BytesIO(code.encode("utf-8")).readline))

new_tokens = []
for token in tokens:
   if token.string == "+":
       token = tokenize.TokenInfo(string="*", start=token.start, end=token.end, line=token.line, type=token.type)
   new_tokens.append(token)

new_funct: str = tokenize.untokenize(new_tokens)

Como se puede observar, el proceso de tokenizar el código produce una serie de tokens que se componen de tipo, string y posición. Deben ser consistentes (es decir, que las posiciones sucesivas deben cuadrar), por lo que introducir cambios en algún punto podría implicar modificar el resto de tokens sucesivos.

Análisis sintáctico

El siguiente paso sería el análisis sintáctico. En este paso se genera un Abstract Syntax Tree, y para ello usamos el módulo ast con el que podremos obtener el comportamiento de nuestro código modelado con clases.

import ast, inspect

def suma(a: int, b: int) -> int:
   return a + b

ast_suma = ast.parse(inspect.getsource(suma))
ast_suma.body[0].body[0].value.op = ast.Mult()

new_code = compile(ast_suma, filename="<ast>", mode="exec")
exec(new_code)

assert suma(10, 5), 50

Cuando queremos analizar el código, cuanto más cerca de los primeros pasos del compilador nos encontramos más compleja es la edición de código. En los ejemplos anteriores, tenemos 24 tokens pero solo 10 clases.

Otro de los inconvenientes de usar el proceso del compilador es que, para analizar el código, debemos usar la misma versión de Python que estamos analizando.

import ast, tokenize
from io import BytesIO

py310 = 'def use_match(status):\n  match status:\n    case "ok":\n      return "All right"\n    case _:\n      return "Something\'s wrong"\n'

tokens = list(tokenize.tokenize(BytesIO(py310.encode("utf-8")).readline))
print([t.string for t in tokens if t.type == tokenize.ERRORTOKEN])
ast.parse(py310)

Si ejecutamos este ejemplo de código en una versión inferior a Python 3.10, veremos un error.

Analizando nuestro propio código

Como se puede observar en los códigos de ejemplo, tanto para generar el ast como para tokenizar usamos el código en modo texto. Para poder acceder a él, hemos usado la librería “Inspect”. Con esta librería podemos acceder a información del código que está cargado en tiempo de ejecución.

import inspect

def suma(a: int, b: int) -> int:
   return a + b

sig: inspect.Signature = inspect.signature(suma)
return_class = sig.return_annotation
is_async = inspect.iscoroutinefunction(suma)

for param in sig.parameters.values():
   print(param.name, param.annotation)

Con Inspect no podemos modificar el comportamiento de nuestro código, pero nos facilita mucho conocerlo en tiempo de ejecución, permitiéndonos crear un comportamiento basado en él. Por ejemplo, autodocumentación como hace FastAPI.

Basándonos en cómo FastAPI hace uso de la introspección y decoradores que nos brindan tanto el core de Python como las librerías httpx, opentelemetry y pydantic, desde el equipo de Backend decidimos empezar a desarrollar una herramienta que nos permitiese:

• Simplificar el proceso de creación de clientes API.
• Usar tipado de las estructuras de datos para minimizar errores.
• Crear clientes asíncronos para mejorar la performance.
• Crear clientes síncronos para dar soporte al código heredado.
• Generar un código que sea autoexplicativo.
• Mantener la trazabilidad.

Además de todo esto, nos interesaba facilitar el uso de buenas prácticas gracias a las mejoras que los IDEs nos proporcionan por el hecho de tipar nuestro código.

Nada mejor para entender de lo que hablamos que viendo código. Para ello, usaremos la API de ejemplo de petstore.swagger.io. Primero, nos vemos obligados a crear las clases/serializadores de los tipos de entrada y salida.

from enum import Enum
from pydantic import BaseModel
import lima_api

class Status(str, Enum):
   AVAILABLE = "available"
   PENDING = "pending"
   SOLD = "sold"

class Item(BaseModel):
   id: int
   name: str | None = None

class Pet(BaseModel):
   id: int
   category: Item | None = None
   name: str | None = None
   photoUrls: list[str] | None = None
   tags: list[Item] | None = None
   status: Status | None = None

class PetUpdate(BaseModel):
   name: str | None = None
   status: Status | None = None

class PetNotFoundError(lima_api.LimaException):
   ...

class InvalidSupplierError(lima_api.LimaException):
   ...

class ValidationError(lima_api.LimaException):
   ...

Como vemos en el ejemplo, tenemos:

• un enumerado Status, que usaremos para filtrar en /pet/findPetsByStatus,
• los modelos de Item y Pet, que representan a las mascotas que recibiremos en /pet/{petId},
• en el PUT a /pet, el modelo PetUpdate para el POST a /pet/{petId}
• y, por último, los errores esperados según el caso.

Ahora, tendremos que crear el cliente y sus distintas funciones:

class PetStoreClient(lima_api.SyncLimaApi):
   def __init__(self, **kwargs):
       kwargs.setdefault("response_mapping", {
           404: PetNotFoundError,
           400: InvalidSupplierError,
       })
       super().__init__(**kwargs)   

En este caso, estamos definiendo ciertas excepciones para los códigos de error que serán comunes para la mayoría de los endpoints.

A continuación, definimos la función para la búsqueda por estado:

@lima_api.get(
   "/pet/findByStatus",
   response_mapping={
       400: ValidationError,
   },
)
def find_by_status(self, *, status: list[Status]) -> list[Pet]:
   ...

Como podemos ver en el decorador, vamos ha hacer un GET a la url /pet/findPetsByStatus. Los datos que se enviarán como query params serán los de status y la respuesta se procederá por el serializador de Pet y será devuelto. En este caso, para el error http 400, la excepción que queremos recibir sería un ValidationError, al contrario de los flujos por defecto, que sería un InvalidSupplierError.

La siguiente petición será la de obtener los datos de una mascota. En este caso, el nombre del parámetro debe llamarse como el campo que aparece con llaves en la url, y la librería se encargará de sustituir este parámetro en el path.

@lima_api.get("/pet/{pet_id}")
def get(self, *, pet_id: int) -> Pet:
   ...

En la siguiente función hacemos una petición PUT con el contenido del Pet como body en formato Json y, para el código http 405, lanzaremos la excepción ValidationError.

@lima_api.put(
   "/pet/",
   response_mapping={
       405: ValidationError,
   },
)
def full_update(self, *, pet: Pet) -> Pet:
   ...

En ocasiones necesitamos cambiar el naming de la función con respecto al dato que enviamos al servidor. Para conseguirlo, utilizaremos el campo “alias” de las clases de lima_api.parameters.LimaParameter; y con BodyParameter, QueryParameter o PathParameter indicamos dónde vamos a enviar esta información.

@lima_api.post(
   "/pet/{petId}",
   headers={
       "content-type": "application/x-www-form-urlencoded"
   }
)
def update(
   self,
   *, 
   pet_id: int = lima_api.PathParameter(alias="petId"),
   pet: PetUpdate = lima_api.BodyParameter(),
) -> None:
   ...

También necesitamos inicializar el cliente apuntando al servidor que queramos, permitiendo de ese modo apuntar a distintos servidores según el entorno.

sync_client = PetStoreClient(base_url="https://petstore.swagger.io/v2")

Para mejorar el performance, queremos que las distintas peticiones se realicen sobre una misma conexión en lugar de tener que abrir una conexión por petición. Por tanto, es necesario abrir la conexión antes de hacer cualquier petición.

Esto se puede conseguir llamando directamente a los métodos de apertura y cierre:

sync_client.start_client()
...
sync_client.stop_client()

O mediante la cláusula with:

with sync_client:
    ...

En este punto, pensaréis: ¿dónde está la asincronía? Para hacer nuestro cliente asíncrono solo tenemos que cambiar dos cosas:

1. La clase de la que hereda, pasando de lima_api.SyncLimaApi a lima_api.LimaApi
2. Añadir async a cada uno de los métodos que hemos definido anteriormente.

El diff sería tal que así:

@@ -42,7 +42,7 @@ class ValidationError(lima_api.LimaException):
     ...

-class PetStoreClient(lima_api.SyncLimaApi):
+class PetStoreClient(lima_api.LimaApi):
     def __init__(self, **kwargs):
         kwargs.setdefault("response_mapping", {
             404: PetNotFoundError,
@@ -56,11 +56,11 @@ class PetStoreClient(lima_api.SyncLimaApi):
             400: ValidationError,
         },
     )
-    def find_by_status(self, status: list[Status]) -> list[Pet]:
+    async def find_by_status(self, status: list[Status]) -> list[Pet]:
         ...

     @lima_api.get("/pet/{pet_id}")
-    def get(self, pet_id: int) -> Pet:
+    async def get(self, pet_id: int) -> Pet:
         ...

     @lima_api.put(
@@ -69,7 +69,7 @@ class PetStoreClient(lima_api.SyncLimaApi):
             405: ValidationError,
         },
     )
-    def full_update(self, *, pet: Pet) -> Pet:
+    async def full_update(self, *, pet: Pet) -> Pet:
         ...

     @lima_api.post(
@@ -78,7 +78,7 @@ class PetStoreClient(lima_api.SyncLimaApi):
             "content-type": "application/x-www-form-urlencoded"
         }
     )
-    def update(
+    async def update(
         self,
         pet_id: int = lima_api.PathParameter(alias="petId"),
         pet: PetUpdate = lima_api.BodyParameter(),
@@ -91,7 +91,7 @@ class PetStoreClient(lima_api.SyncLimaApi):
             "content-type": "application/x-www-form-urlencoded"
         }
     )
-    def other_update(
+    async def other_update(
         self,
         pet_id: int,
         name: str | None = None,

Conclusiones

A pesar de que el tipado en Python son meramente anotaciones (es decir, no implican necesariamente que el tipo del dato se cumpla), poder usar tipado no solo nos permite mejorar la compresión y el análisis de nuestro código por “type checkers”, sino que además, gracias a un decorador que analiza la definición de la función haciendo uso de Inspect, podemos dotar de lógica adicional a nuestro código.

Si os ha parecido interesante la librería que os hemos enseñado, podéis hacer uso de ella en nuestro github donde, además, podréis ver todos los detalles de implementación o instalarla usando “pip install lima-api”