El otro día publicaba un tweet a raíz de un refactor que hice un código digamos “mejorable”, que yo mismo había escrito días atrás xD, y me gustaría hablar sobre ello.

¿Qué es la inyección de dependencias y porque debemos usarla?

La inyección de dependencias es un patrón de diseño en el que los colaboradores que necesita una clase para funcionar no son creados internamente, sino que son suministrados desde fuera vía constructor.

Este patrón nos va a permitir por una parte no acoplarnos a un comportamiento y además que desde fuera de la clase podamos cambiar cierto comportamiento para casos especiales como pueden ser nuestros tests.

Caso de real de código acoplado

Imaginemos que estamos trabajando y de pronto salta una alerta por un fallo en producción, después del pánico inicial miramos los logs y unos minutos de investigación más tarde descubrimos el problema.

Al parecer se ha generado una excepción de tipo HttpError tratando de leer los usuarios en Google.

La forma de acceder a estos usuarios se hace a través de un repositorio llamado SdkGoogleUsersRepository el cual, como su propio nombre indica, hace uso de la librería de Pyhton para comunicarse con Google.

Veamos que pinta tiene el repositorio:

class SdkGoogleUsersRepository(GoogleUsersRepository):
    def find_all_users(self) -> List[GoogleUser]:
        directory_service = self._generate_directory_service()
        users = self._find_all_users(directory_service)
        return self._generate_google_users(users)

    @staticmethod
    def _generate_directory_service() -> Resource:
        scopes = ["https://www.googleapis.com/auth/admin.directory.user"]
        credentials = os.getenv(GOOGLE_SERVICE_ACCOUNT_CREDENTIALS)
        creds = service_account.Credentials.from_service_account_info(
            credentials, scopes=scopes, subject="my-email@abc.xz"
        )
        return build("admin", "directory_v1", credentials=creds)

    @staticmethod
    def _find_all_users(directory_service: Resource) -> List[Dict]:
        request = directory_service.users().list(customer="my_customer", orderBy="email")
        response = request.execute() !! # THE PROBLEM IS HERE!!!
        return response.get("users", [])

    def _generate_google_users(self, users: List[Dict]) -> List[GoogleUser]:
        ...

class SdkGoogleUsersRepositoryFactory:
    @staticmethod
    def make() -> GoogleUsersRepository:
        return SdkGoogleUsersRepository()

¿Según los logs el error se ha generado en la línea response = request.execute() por lo que ya solo nos quedaría crear un test que nos permita replicar este error y corregirlo no?

¿NO?

Pues siento deciros que no, ojalá fuera posible, pero ahora mismo hacer ese test tan sencillo es un drama.

¿Y sabéis por qué es un drama? Porque nuestro repositorio está acoplado completamente a la librería debido a que la instanciación se genera internamente, por lo que es imposible cambiar su comportamiento desde fuera haciendo uso de dobles.

Pero bueno cero dramas, todo en esta vida tiene solución menos la muerte!.

Desacoplemos nuestro repositorio

Como en todos mis posts anteriores vamos a hacer TDD así que empecemos por el test que nos gustaría tener:

class TestSdkGoogleUsersRepository:
    def test_raise_an_exception_finding_users(self) -> None:
        dummy_directory_service = TestData.a_google_users_dummy_directory_service()
        user_repository = SdkGoogleUsersRepository(dummy_directory_service)

        expect(lambda: users_repository.find_all_users()).to(raise_error(GoogleServerErrorException))

En el test necesitamos un discovery_service y hace poco hablamos del concepto Object Mother así que vamos a aplicarlo aquí:

    @staticmethod
    def a_google_users_dummy_directory_service() -> "GoogleUsersDummyDirectoryService":
        return GoogleUsersDummyDirectoryService()

A su vez nuestra Object Mother hará uso de un builder del que también hablamos hace poco.

Esta parte es la más tricky, ya que vamos a aprovechar la flexibilidad de Python para hacernos una clase a medida que replique el comportamiento de la librería de Google. Como hemos podido ver en nuestro test la idea es poder inyectarlo en el constructor de nuestro SdkGoogleUsersRepository.

Estudiando un poco nuestro SdkGoogleUsersRepository podemos ver como una vez creado el discovery_service en el método generate_directory_service se hacen las siguientes llamadas:

request = directory_service.users().list(customer="my_customer", orderBy="email")
response = request.execute()

Es decir necesitamos una clase con un método users que nos retorne un objeto que tenga un método list con una par de parámetros customer y orderBy, que a su vez retornará otro objeto que tenga un método execute.

Lo sé, sé lo que estás pensando, pero así como se comporta la librería.

Si conseguimos replicar este comportamiento podremos inyectar esta clase a nuestro repositorio y reproducir el error de producción. Manos a la obra:

class GoogleUsersDummyDirectoryService:
    @staticmethod
    def execute() -> Dict:
        raise GoogleServerErrorException

    def list(self, customer: str, orderBy: str) -> "GoogleUsersDummyDirectoryService":
        return self

    def users(self) -> "GoogleUsersDummyDirectoryService":
        return self

Por último tendríamos que actualizar nuestro repositorio SdkGoogleUsersRepository moviendo la parte de la configuración fuera de la clase y que esta reciba un discovery_service ya listo para usar.

Además de esto debemos introducir un try/except para controlar la excepción HttpError y levantar una propia de como puede ser GoogleServerErrorException.

Aquí es donde se produce la inyección que nos permitirá cambiar el comportamiento del repositorio desde los tests

class SdkGoogleUsersRepository(GoogleUsersRepository):
    def __init__(self, directory_service: Resource) -> None:
        self.directory_service = directory_service

    def find_all_users(self) -> List[GoogleUser]:
        users = self._find_all_users()
        return self._generate_google_users(users)

    def _find_all_users(self) -> List[Dict]:
        request = self.directory_service.users().list(customer="my_customer", orderBy="email")
        try:
            response = request.execute()
            return response.get("users", [])
        except HttpError:
            raise GoogleServerErrorException("Unable to connect to Google")

    def _generate_google_users(self, users: List[Dict]) -> List[GoogleUser]:
        ...

class SdkGoogleUsersRepositoryFactory:
    @staticmethod
    def make() -> GoogleUsersRepository:
        scopes = ["https://www.googleapis.com/auth/admin.directory.user"]
        credentials = os.getenv(GOOGLE_SERVICE_ACCOUNT_CREDENTIALS)
        creds = service_account.Credentials.from_service_account_info(
            credentials, scopes=scopes, subject="my-email@abc.xz"
        )
        directory_service = build("admin", "directory_v1", credentials=creds)
        return SdkGoogleUsersRepository(directory_service)

¡Si lanzamos nuestro test deberíamos verlo en verde!

Conclusiones

Creo que hemos podido ver un ejemplo muy visual de porque un software acoplado es un problema y como hacer uso de la inyección nos permite testear nuestro código de una forma supersencilla.

Aún podríamos ir más allá y no quedarnos en la inyección si no que además podrías aplicar inversión de dependencias, pero eso para otro post!.

!Un saludo!