Forem: Yoandy Rodriguez Martinez

Workshop de Golang

Yoandy Rodriguez Martinez — Sat, 06 Feb 2021 02:01:15 +0000

¡Hola Gophers!

He compilado un conjunto de ejemplos y diapositivas para aprender Go. Si quieres aprender, colaborar o conoces a alguien interesado puedes darte una vuelta por el repo en github. Si gusta no olvide dejar su ⭐️.

Pipelines en Rust (I)

Yoandy Rodriguez Martinez — Thu, 24 Dec 2020 18:27:39 +0000

Pipeline es un patrón de diseño muy útil cuando tienes datos que
deben ser procesados en una secuencia de etapas donde cada etapa toma
como entrada la salida de la anterior. En cierta manera un
pipeline es similar a componer funciones pero el nivel de
complejidad es mucho más elevado debido a factores como backpressure, deadlocks o cancelación.

Go es un lenguaje especialmente capacitado para programar
pipelines debido a sus características especiales para el manejo de errores y concurrencia. Pero ¿cómo sería usar pipelines en Rust?

En este post vamos a:

Definir las estructuras necesarias para crear pipelines.
Hacer uso del sistema de tipos del lenguaje para nuestras ventajas.
Hablar un poco de concurrencia usando hilos.

Paso 1: El `trait Step`

En el mejor espíritu de Rust hagamos un trait que represente la
capacidad de formar parte de un pipeline. Vamos a llamarle Step

pub trait Step {
    type Item;
    pub run(&self, it: Self::Item) -> Self::Item
}

Hecho, hasta la próxima. O... mejor aún, miremos más de cerca la
definición de Step.

Es un trait bastante sencillo. Tiene un tipo asociado Item y una
función run que acepta y retorna Item. La forma en que la función
está especificada no permite que Item sea una referencia. Y
finalmente it no es mutable, por lo que el parámetro de entrada es
consumido por la función y el valor de retorno es generado por ella.

Step es muy simple de implementar, veamos un ejemplo:

pub struct Multiplier {
    value: u8,
}

impl Step for Multiplier {
    type Item = u8;
    fn run(&self, it: u8) -> u8 {
        return self.value * it;
    }
}

Podemos usar Multiplier para crear pasos que... bueno, multipliquen
su valor de entrada por un número dado.

let by2 = Multiplier{value:2};
println!("Multiplicado por 2 {0}", by2.run(5)) // 10

Paso 2: Pipeline

Ahora solo tenemos que encadenar los pasos para formar un pipeline.
Ya que tenemos un número variable de pasos y todos implementan el
mismo trait, podemos guardarlos en un vector de trait objects

pub struct Pipeline<T> {
    v: Vec<Box<dyn Step<Item = T>>>,
}

La implementación de Pipeline es extremadamente corta.

impl<T> Pipeline<T> {
    fn new() -> Pipeline<T> {
        Pipe { v: Vec::new() }
    }

    fn add(&mut self, x: impl Step<Item = T> + 'static) {
        self.v.push(Box::new(x));
    }
}

Un detalle en add: para adicionar un Step, debemos asegurarnos que
viva lo suficiente¹, por lo que indicamos con 'static. Por
cuestiones de estilo (y pensando en el futuro) podemos hacer que Pipeline se comporte como cualquier otro Step.

impl<T> Step for Pipeline<T> {
    type Item = T;
    fn run(&self, it: T) -> T {
        todo!()
    }
}

Listo, tenemos la capacidad de hacer subpipelines y sólo nos ha
costado unas líneas. La operación dentro de run es tan simple como hacer un fold.

fn run(&self, it: T) -> T {
    self.v.iter().fold(it, |acc, x| x.run(acc))
}

Hagamos un pequeña prueba²:

#[test]
fn test_pipeline() {
    let mut p = Pipeline::<u8>::new();
    p.add(Multiplier { value: 2 });
    p.add(Multiplier { value: 5 });
    assert_eq!(p.run(10),100);
}

Paso 3: Pipelines en el mundo real.

Nuestras pipelines funcionan bastante bien en el mundo de las
multiplicaciones de números pequeños, pero en el resto de los mundos existe algo llamado "errores" y con la definición actual (por muy elegante que sea) no tenemos modo de detectar si uno de los pasos falla. Es hora de sacar Result

use std::error::Error;

type StepResult<T> = Result<T,Box<dyn Error>>;

trait Step {
    type Item;
    fn run(&self, it: Self::Item) -> StepResult<Self::Item>;
}

impl Step for Multiplier {
    type Item = u8;
    fn run(&self, it: u8) -> StepResult<u8> {
       Ok(self.value * it)
    }
}

Tener Box<dyn Error> nos da la garantía de poder manejar errores de cualquier tipo. La implementación de Pipeline debe tener esto en cuenta y propagar el estado de error hasta el resultado final.

impl<T> Step for Pipeline<T> {
    type Item = T;
    fn run(&self, it: T) -> StepResult<T> {
        self.v.iter().fold(Ok(it), |acc, x| acc.and_then(|v| x.run(v)))
    }
}

Hora de ajustar nuestra prueba.

#[test]
fn test_pipeline_ok() {
    let mut p = Pipeline::<u8>::new();
    p.add(Multiplier { value: 2 });
    p.add(Multiplier { value: 5 });
    assert_eq!(p.run(10),Ok(100));
}

Y crear una nueva para cuando algún Step falla.

struct ErrStep;

impl Step for ErrStep {
    type Item = u8;

    fn run(&self, it:u8) -> StepResult<u8> {
       Err("This will fail")?
    }
}

fn test_pipeline_ok() {
    let mut p = Pipeline::<u8>::new();
    p.add(Multiplier { value: 2 });
    p.add(ErrStep{});
    assert!(p.run(10).is_err());
}

Pensando en paralelismo y concurrencia.

Recuento:

Tenemos la posibilidad de hacer un Pipeline compuesto de distintas implementaciones de Step.
Tenemos la forma de propagar errores en el Pipeline

El próximo paso natural sería intentar usar nuestro diseño actual para ejecutar tareas en paralelo. Por desgracia, aún no hemos llegado a ese punto.

Una de las ventajas de Rust es la garantía de que el compilador va a detectar problemas comunes de seguridad de hilos (ejemplo, acceder desde dos hilo distintos a la misma zona de memoria), para esto el la biblioteca estándard incluye marcadores como Send y Sync, o tipos especiales como Arc.

Adicionalmente, en el espíritu de compartir comunicando, los datos entre implementaciones de Step deberían pasar usando canales o colas concurrentes, esto ayudaría también con otros aspectos que mencionamos al inicio del artículo (como backpressure) pero que no tratamos por no ser necesarios para una implementación secuencial.

The end.

Con todos los puntos del plan cumplidos, me retiro hasta la próxima aventura. Mientras tanto si estás interesado en el tema de pipelines en Rust recomiendo mirar pipelines o rayon, ambas con implementaciones muy interesantes.

Aunque implementar Step para una referencia o un tipo con
restricciones de tiempo de vida no es trivial, tampoco es
imposible. ↩
Las autoridades advierten que hacer pruebas unitarias
después de escribir funcionalidades es malo para la salud. ↩

Emacs + Hugo: Haciendo hugo-blog-mode (II)

Yoandy Rodriguez Martinez — Tue, 06 Oct 2020 21:45:15 +0000

En el artículo anterior (hace unos cuantos meses ya) estuve hablando sobre como comencé
automatizando el flujo de publicación del blog usando scripts. En este les mostraré como llevamos esos scripts a un modo de Emacs.

Manos al Emacs.

Una de las ventajas de Emacs es su extensibilidad, con un poco de
código Elisp podemos adicionar nuevas funcionalidades al editor.
Aunque sería posible hacer un modo¹ para crear toda una experiencia relacionada con el manejo de blogs, es mucho más sencillo crear comandos².

Planeando las funcionalidades.

Necesitamos 3 funcionalidades básicas para trabajar en nuestro blog.

Crear nuevos artículos.
Publicar el blog en GitHub Pages.
Visualizar en el entorno local cualquier cambio que se produzca.

Con esto en mente podemos crear el esqueleto de las funciones

(defun hugo-blow-new (path)
    "Create new content in PATH.")
(defun hugo-blog-publish ()
    "Generate the site and commit everything.")
(defun hugo-blog-preview (arg)
    "Launches a preview HTTP server. If ARG is provided also render drafts.")

Opciones de configuración.

Vamos a necesitar algunas opciones configurables:

El nombre (o camino) del ejecutable de hugo.
La raiz del proyecto del blog.
El nombre del buffer que usaremos para interactuar con el comando.

(defgroup hugo-blog nil
  "Hugo blog mode customizations"
  :group 'tools)

(defcustom hugo-blog-command "hugo"
  "Path to hugo's executable."
  :group 'hugo-blog
  :type 'string)

(defcustom hugo-blog-project ""
  "Blog directory project."
  :group 'hugo-blog
  :type 'string)

(defcustom hugo-blog-process-buffer "*hugo-blog-process*"
  "Hugo blog process buffer."
  :group 'hugo-blog
  :type 'string)

Funciones de soporte.

Definidos nuestros puntos de entrada y las opciones configurables
necesitamos englobar algunas funcionalidades comunes en funciones de soporte.

Emacs nos brinda funciones para ejecutar programas externos o comandos del shell. La forma simple sería ejecutar los scripts
de shell que tenemos definidos, pero es mucho más interesante reproducir la funcionalidad en Elisp. Veamos por ejemplo como podemos ejecutar hugo y capturar la salida en busca de mensajes de error.

(defun hugo-blog-run-command (command parameters)
  "Run COMMAND with PARAMETERS with `hugo-blog-project' as working directory.
Returns the command's output as a string"
  (cd hugo-blog-project)
  (let ((output (shell-command-to-string
                 (concat hugo-blog-command
                         " "
                         command
                         " "
                         parameters))))
    (if (string-match-p "Error" output)
        nil
      output)))

La función retorna nil en caso de error o el mensaje resultado de la llamada en otro caso

Así mismo podemos definir el resto de las funciones necesarias.

(require 'git) ;; git related functions

(defmacro with-git-repo (repo &rest body)
  "Run BODY using git repository REPO."
  `(let ((git-repo ,repo))
     ,@body))

(defun git-modified-files ()
  "Return list of untracked files."
  (git--lines
   (git-run "ls-files" "-m" "--exclude-standard")))

(defsubst hugo-blog-submodule ()
  "Inline function to get the submodule."
  (concat hugo-blog-project (f-path-separator) "public"))

(defun hugo-blog--commit-all ()
  "Commits the submodule and then the project."
  (with-git-repo  (hugo-blog-submodule)
                 (when (git-modified-files)
                   (git-add)
                   (git-commit (concat "Commit on "
                                       (current-time-string)))))
  (with-git-repo  hugo-blog-project
                 (when (git-modified-files)
                   (git-add)
                   (git-add "public") ;; Let's be really sure
                   (git-commit (concat "Commit on "
                                       (current-time-string))))))

Completando las funcionalidades

Con ayuda de las funciones auxiliares podemos desarrollar los comandos necesarios. Es importante notar que todas las funciones están marcadas como autoload e interactive así estarán disponibles como comandos en el editor y sería más fácil distribuirlas en un paquete.


;;;###autoload
(defun hugo-blog-new (path)
  "Create new content in PATH."
  (interactive "sNew content path: ")
  (cd hugo-blog-project)
  (let ((output (hugo-blog-run-command "new" path)))
    (if output
        (find-file-existing  (car (split-string output " ")))
      (error "Command hugo returned an error, check your configuration"))))

;;;###autoload
(defun hugo-blog-publish ()
  "Generate the site and commit everything."
  (interactive)
  (save-some-buffers) ;; avoid commiting emacs weird files
  (when (yes-or-no-p "This will commit changes, are you sure? ")
    (hugo-blog-run-command "--noChmod" "--noTimes --ignoreCache")
    (hugo-blog--commit-all)))

;;;###autoload
(defun hugo-blog-preview (arg)
  "Launches a preview HTTP server. If ARG is provided also render drafts."
  (interactive "P")
  (unless (process-status "hugo")
    (cd hugo-blog-project)
  (when arg
    (start-process "hugo" hugo-blog-process-buffer
                   hugo-blog-command "-D" "server"))
  (unless arg
    (start-process "hugo" hugo-blog-process-buffer
                   hugo-blog-command "server")))
  (sleep-for 5)
  (with-current-buffer hugo-blog-process-buffer
    (goto-char (point-max))
    (if (re-search-backward "http://localhost:[0-9]+/" nil t)
        (browse-url (match-string 0))
      (error "Error executing hugo"))))

Los modos de son conjuntos de funcionalidades que extienden el editor en distintas maneras, muchos incluso hacen cambios en la forma en que interactuamos con Emacs ↩
Los comandos son funcionalidades específicas que no tienen por que estar relacionadas con ningún modo en particular. ↩

Microservicios en .NET Core 3.1

Yoandy Rodriguez Martinez — Fri, 06 Mar 2020 20:06:47 +0000

La nueva versión LTS de .NET Core trae consigo muchas ventajas para ellos desarrolladores de microservicios. Una de las más esperadas es laestabilización de la API para ejecutar servicios en segundo plano o_workers_ utilizando Microsoft.Extensions.Hosting

¿Qué son los workers?

Si eres desarrollador de ASP.NET Core debes estar familiarizado con elpaquete Microsoft.AspNetCore.Hosting y las clasesWebHostBuilder yWebHost.

public class Program
{
    public static void Main(string[] args)
    {
       var config = new ConfigurationBuilder()
            .AddCommandLine(args)
            .AddJsonFile("hosting.json", optional: true)
            .AddEnvironmentVariables()
            .Build();
      var host = new WebHostBuilder()
           .UseKestrel()
           .UseContentRoot(Directory.GetCurrentDirectory())
           .UseConfiguration(config)
           .Build();

       host.Run();
    }
 }

WebHost es responsable de iniciar la configuración, manejar el ciclode vida de los servicios, mantener el contenedor de dependencias ycomunicarse/iniciar el servidor web (Kestrel en este caso).

WebHost también nos brinda la posibilidad de iniciar servicios ensegundo plano utilizando IHostedService.

Si no estás familiarizado con el tema, puedes informarte más con este excelente post

Microservicios en aplicaciones de consola.

El nuevo host genérico (HostBuilder) no requiere de la maquinaria deASP.NET Core, lo cual hace posible crear aplicaciones de consola quehagan uso de las bondades a las que estamos adaptados.

Veamos un ejemplo:

 public class Program
    {
        public static void Main(string[] args)
        {
            try {
            CreateHostBuilder(args).Build().Run();
            }
            catch(OperationCanceledException ex)
            {
                Console.WriteLine("-----Terminating-----");
            }
        }

        public static IHostBuilder CreateHostBuilder(string[] args) =>
            Host.CreateDefaultBuilder(args)
                .ConfigureAppConfiguration((hostingContext, config) =>
                {
                    var env = hostingContext.HostingEnvironment.EnvironmentName;
                    var builder = new ConfigurationBuilder()
                            .SetBasePath(Directory.GetCurrentDirectory())
                            .AddJsonFile("settings.json")
                            .AddEnvironmentVariables()
                            .Build();
                })
                .ConfigureServices((hostingContext, services) =>
                {

                    // AWS Configuration
                    var options = hostingContext.Configuration.GetAWSOptions();
                    services.AddDefaultAWSOptions(options);
                    services.AddAWSService<IAmazonSQS>();

                    // Worker Service
                    services.AddHostedService<Worker>();
                    services.AddLogging();

                }).ConfigureLogging((hostcontext, configLogging) =>
                {
                    configLogging.AddConsole();
                    configLogging.AddDebug();
                });
    }

El programa anterios hace uso del host genérico para:

Cargar la configuración externa (variables de entorno y archivo json).
Inicializar un servicio para acceder a SQS.
Configurar logs para la consola.
Crear un servicio en segundo plano.

La clase Worker hereda de BackgroundService y es la encargada deejecutar la lógica de nuestro microservicio

public class Worker: BackgroundService {

     public Worker(ILogger<Worker> logger, IAmazonSQS sqs)
        {
            _logger = logger;
            _sqs = sqs;
        }

     protected async Task ExecuteAsync(int number, CancellationToken cancel)
       {
            while (!cancel.IsCancellationRequested)
            {
                // Procesar mensajes desde SQS
            }
       }
}

Integración con el sistema operativo

Si desarrollas servicios monolíticos o para software legado, puedeshacer uso del host genérico para integrar tu aplicación al sistemaoperativo.

Para integrarse como servicio a Windows se puede utilizar lasextensiones en Microsoft.Extensions.Hosting.WindowsService parainicializar nuestro programa.

 Host.CreateDefaultBuilder(args)
        .UseWindowsService()

En el casos de Linux sólo están soportadas las distribuciones que incluyenSystemd

  Host.CreateDefaultBuilder(args)
        .UseSystemd()

awstail: AWS Logs a la vieja usanza

Yoandy Rodriguez Martinez — Mon, 20 Jan 2020 16:50:59 +0000

awstail es una herramienta muy simple que cumple dos funciones:

Darme la excusa para aprender Rust.
Monitorear un grupo de logs en AWS Cloudwatch desde la terminal.

Por el momento, estas son las opciones que permite la herramienta:

awstail 0.3.0
Yoandy Rodriguez <yoandy.rmartinez@gmail.com>
like tail -f for AWS Cloudwatch

USAGE:
    awstail.exe [OPTIONS] <LOG_GROUP>

FLAGS:
    -h, --help Prints help information
    -V, --version Prints version information

OPTIONS:
    -p <PROFILE> Profile if using other than 'default'
    -r <REGION> AWS region (defaults to us-east-1)
    -s <SINCE> Take logs since a given time (defaults to 5 minutes)
    -w <WATCH> Keep watching for new logs every n seconds (defaults to 10)

ARGS:
    <LOG_GROUP> Log group name

Posibles opciones futuras:

Agregar logs de más de un grupo.
Color diferenciado para mensajes de error.
Listar grupos de logs existentes.

Cola de tareas en Python (I)

Yoandy Rodriguez Martinez — Mon, 04 Nov 2019 18:01:46 +0000

Después de unos meses trabajando en Go se llegan a extrañar las abstracciones del lenguaje para concurrencia. Hoy por ejemplo necesitaba hacer una cola de tareas en Python utilizando AsyncIO. En Go esto sigue una estructura sencilla:

// Este es la gorutina que procesa los trabajos
func worker(jobChan <-chan Job) {
    for job := range jobChan {
        process(job)
    }
}
// Creamos un canal a donde enviar los datos
jobChan := make(chan Job, 10)
// Arrancamos la gorutina
go worker(jobChan)
// Enviamos datos para un trabajo, esto puede ser desde cualquier
// gorutina
jobChan <- job
//Indicamos que ya no vamos a procesar más datos
close(jobChan)

Veamos si podemos lograr un equivalente en Python:

# Primer intento
import asyncio
async def worker(queue):
    while True:
        process(await queue.get())
        # indicamos que se procesó el trabajo
        queue.task_done()

# Creamos la cola de mensajes
queue = asyncio.Queue(10)
# arrancamos el worker
worker_task = asyncio.create_task(worker(queue))
# Enviamos mensajes a la cola
await queue.put(x)

Hasta acá todo parece bien, pero: ¿Cómo indicamos que no se van a procesar más datos?

En Go la operación close sobre un canal hace que el ciclo termine después de procesar cualquier elemento pendiente. En Python a primera vista podemos cancelar la tarea creada enworker_task pero eso nos deja con la posibilidad de que varias tareas se queden en la cola. Por otro lado, la corutina join() de asyncio.Queue nos permite esperar a que ya no existan elementos en la cola, aunque no garantiza que worker se detenga.

La solución es utilizar una mezcla de los dos:

await queue.join() # Esperar a que se procesen los pendientes
worker_task.cancel() # Cancelar la tarea

A QA engineer enters a bar

Yoandy Rodriguez Martinez — Wed, 03 Jul 2019 16:42:08 +0000

A QA Engineer enters a bar:

Stares at the barman for twenty minutes.
Asks for two beers and a straw.
Asks for sparkling water with lemon juice.
Asks for a unicorn, a lizard and some tequila.

Github bloquea acceso desde Cuba

Yoandy Rodriguez Martinez — Mon, 03 Jun 2019 15:36:08 +0000

Bueno, GitHub a decidido que el código fuente de mis proyectos no debe ser visto desde el país donde vivo y los produzco. Muchas gracias por golpear aún más a los que ya tenemos pocas puertas abiertas y ninguna opinión en el proceso.

Ayúdame a protestar en Twitter

UPDATE

Parece que han restaurado el acceso, no hay comunicados oficiales todavía.

Emacs + Golang. Edición 2019

Yoandy Rodriguez Martinez — Mon, 25 Mar 2019 19:34:27 +0000

Go 1.12 fue lanzado hace un mes y entre los cambios más significativos se encuentran varias mejoras en el soporte para módulos (que va a ser el método por defecto en la versión 1.13).

Si eres usuario de Emacs y te interesa desarrollar utilizando Go, aquí tienes una guía sencilla de como habilitar el soporte para el lenguaje.

Paso 1. Language server protocol

Si tienes una versión actualizada de lsp-mode ya tienes soporte para bingo, una herramienta que provee un servidor de lenguajes para Go y que tiene soporte integrado para trabajar con módulos

(use-package lsp-mode
  :commands (lsp)
  :config
  (use-package company-lsp
    :config
    (add-to-list 'company-backends 'company-lsp)))

(use-package lsp-ui
  :hook (lsp-mode . lsp-ui-mode)
  :config
  (define-key lsp-ui-mode-map
    [remap xref-find-definitions] #'lsp-ui-peek-find-definitions)
  (define-key lsp-ui-mode-map
    [remap xref-find-references] #'lsp-ui-peek-find-references)
  (setq lsp-ui-sideline-enable nil
        lsp-ui-doc-enable t
        lsp-ui-flycheck-enable nil
        lsp-ui-imenu-enable t
        lsp-ui-sideline-ignore-duplicate t))

Paso 2. Go mode y utilidades

Como lsp-mode es un minor mode necesitamos el soporte para el lenguaje (font-lock entre otras cosas). Los pasos necesarios son:

Instalar goimports
Instalar golint o gofmt.

(use-package flycheck)
(use-package go-mode
  :config
  ; Use goimports instead of go-fmt
  (setq gofmt-command "goimports")
  (flycheck-mode)
  (add-hook 'go-mode-hook 'company-mode)
  ;; Call Gofmt before saving
  (add-hook 'before-save-hook 'gofmt-before-save)
  (add-hook 'go-mode-hook #'lsp)
  (add-hook 'go-mode-hook #'flycheck-mode)
  (add-hook 'go-mode-hook '(lambda ()
                 (local-set-key (kbd "C-c C-r") 'go-remove-unused-imports)))
  (add-hook 'go-mode-hook '(lambda ()
                 (local-set-key (kbd "C-c C-g") 'go-goto-imports)))
  (add-hook 'go-mode-hook (lambda ()
                (set (make-local-variable 'company-backends) '(company-go))
                (company-mode))))

Paso 3. Comenzar a programar en Go.

Con la configuración anterior puedes empezar a programar en Go

Completamiento y referencias cruzadas
Revisión de código mientras escribes.
Formato automático al salvar.
Un efecto aprecido al hover de ese editor taaan popular

Paso 4. Compilar y debug.

Puedes configurar el comando de compilación por defecto que utilizas cuando editas archivos Go.

;; Adicionar en la sección :config de `go-mode`
(add-hook 'go-mode-hook #'setup-go-mode-compile)

;; adicionar en el mismo archivo

(defun setup-go-mode-compile ()
  ; Customize compile command to run go build
  (if (not (string-match "go" compile-command))
      (set (make-local-variable 'compile-command)
           "go build -v && go test -v && go vet")))

Si prefieres los debuggers interactivos debes instalar delve y añadir la siguiente línea en tu archivo de inicio de Emacs.

(use-package go-dlv)

Los comandos dlv y dlv-current-func son los puntos de entrada al debugger.

The Ultimate Web Framework Deathmatch!

Yoandy Rodriguez Martinez — Thu, 14 Mar 2019 18:34:13 +0000

Are those benchmarking blog post bothering you? Tired of listening to people boast about how good framework X is? Undecided wether to use framework X or framework Y in your new startup?. Fear not!!! You can now (easily) launch your own edition of: The Ultimate Web Framework Deathmatch using TechEmpower's Web Framework benchmarks suite.

The project has a Github repo with all the things you need to run your own tests. As this post is beeing written there's a fierce competition going between an aio-libs based python app and net/http with SQLx, both using PostgreSQL, so just read the instructions and let the games begin! You can add your results to the project via a PR or just post your benchmarking as a comment here.

Disclaimer

In case someone is wondering, I have no idea of who the good folks at TechEmpower are, neither I work for them or receive any compensation from them. I just thought the whole Deathmatch thing sounded like a way to have some productive fun.

Plumber: ETL simple para Python

Yoandy Rodriguez Martinez — Tue, 12 Mar 2019 14:14:20 +0000

Implementando una feature para uno de los proyectos en los que trabajo terminé necesitando hacer el típico proceso ETL que tomara las cosas de una fuente X, las pasara por 1 o varios filtros y las cargara en nuestra base de datos documental.

Ahora, si bien existen frameworks de excelente calidad como petl y Bonobo que cubren este campo, me encontré con los siguientes problemas:

Existe mucho código legacy en el proyecto.
Exceptuando la parte de ETL, la complejidad de proceso es bastante alta, incrementarla no sería lo más adecuado.
El componente ETL no se ejecutaría por si mismo sino como parte de una solución.
La mayoría de los datos que agregamos usan formatos no convencionales.

Aunque en apariencia sencillos, estos cuatro puntos me llevaron a crear mipropio miniframework para componentes ETL que es fácil de integrar con código existente y no hace ningún tipo de suposiciones con respecto a la forma de los datos, plumber.

Características de plumber.

Para el desarrollo del framework me basé mucho en mi experiencia anterior con kiba, una de esas joyas que siempre vienen desde la comunidad de Ruby. Kiba hace uso de las bondades de Ruby para crear DSL’s dejando al desarrollador que se centre en los datos de la manera que considere correcta. En plumber esta idea está reflejada en cuatro conceptos:

Extractors (que acceden al origen de datos).
Transformers (que manipulan los datos obtenidos).
Writers (que persisten los datos).
Pipes (que controlan todo el proceso).

Extractors.

Un Extractor es una clase o función que emite datos para una tubería (pipe), los datos pueden tener cualquier origen y cualquier forma, los extractors son independientes del resto del proceso.

Aunque es posible utilizar una función como extractor, lo recomendado en la mayoría de los casos es heredar de la clase plumber.pipe.Extractor e implementar los métodos read, setup y cleanup. Varios extractors se pueden unir para formar uno solo que emita una secuencia (tupla) con los valores decada uno.

Transformers.

Un Transformer es cualquier función marcada con @transformer o una clasederivada de plumber.pipe.Transformer. Es posible unir varios transformers e nuna cadena de procesamiento

Writer.

Finalmente los Writers heredan de plumber.pipe.Writer. Como generalmente acceden a recursos externos, es requerido que implementen los métodos setup ycleanup además de write.

Todos los elementos usan anotaciones PEP 484.

Un ejemplo sencillo:

Uno de los casos de uso que motivaron la creación del framework fue obtener una lista de registros de un archivo con formato propietario. El contenido delarchivo es más o menos el siguiente:

Campo identificador de cliente. Comienza en la posición 0, tiene longitud entre 1 y 20.
Campo identificador de transacción. Comienza en la posición 22. Es numérico y tiene longitud 8.
Campo identificador de transacción relacionada. Comienza en la posición 32.Tiene longitud 8, es opcional y de no estar se ponen espacios en blanco.
Campo motivo de la transacción. Comienza en la posición 42 y longitud entre 10 y 50. Es de tipo alfanumérico, incluye espacios, no tiene delimitaciones.

Veamos un ejemplo de como procesar estos archivos:

from plumber import pipe # API síncrona.

@pipe.extractor
def read_file():
    file_name = os.environ['FILENAME']
    with open(file_name) as f:
        for x in f.readlines():
            yield process_line(x)

@pipe.transformer
def csvfy(element):
    yield ','.join(map(str,element))

class SaveData(pipe.Writer):

    def __init__ (self, filename):
        self.filename = filename

    def setup(self):
        self._file = open(f,'w')

    def cleanup(self):
        self._file.close()

    def write(x):
        self._file.write(x)

tuberia = pipe.Pipe(
    read_file,
    csvfy,
    SaveData("prueba.csv")
)
tuberia.run()

¿Qúe queda por hacer?

Ahora que plumber salió a la luz es hora de convertirlo en un framework ETL medianamente decente. Los próximos pasos son:

Adicionar la posibilidad de hacer Writers como funciones con administración de contexto integrada.
Poder inicializar los Extractors de manera sencilla.
Mejorar el tratamiento de errores.
Concurrencia y/o paralelismo.

Python: Decoradores estándar y su uso

Yoandy Rodriguez Martinez — Wed, 16 Jan 2019 03:11:42 +0000

Entre las baterías includas de Python, vienen varios decoradores que nos facilitan la vida. Aquí una lista de algunos y breve explicación acerca de su uso.

Decoradores para OOP

property

Uno de los más usados. Convierte un conjunto de métodos en un descriptor que hace las funciones de propiedad. Útil cuando queremos adicionar algún tipo de lógica a propiedades de la clase.

class C:
    def __init__ (self):
        self._x = None

    def getx(self):
        return self._x or request.get('http://mydata.com/x')

    def setx(self, value):
        self._x = value
        request.post('http://mydata.com/x', data=x)

    def delx(self):
        del self._x
        request.delete('http://mydata.com/x')

    x = property(getx, setx, delx, "x como proxy a servicio REST.")

staticmethod

Define métodos estáticos¹. Pueden ser llamados tanto desde la clase o una de sus instancias

class C:

    @staticmethod
    def calculo_externo(val, val2):
        return val + val2

Nótese la falta del parámetro implicito inicial.

classmethod

Convierte un método normal a un método de clase. Los métodos de clase reciben la clase como parámetro implícito en vez de una de las instancias. Uno de los casos de uso más comunes es crear formas alternativas de instanciar la clase.

class C:

    def __init__ (self, x, y ,z):
        self.x = x
        self.y = y
        self.z = z

    @classmethod
    def from_dict(cls, data):
        return cls(data[0], data[1], data[3])

abc.abstractmethod

Métodos que tienen que ser implementados obligatoriamente por las subclases. El equivalente a interfaces o clases abstractas que tanto encuentras en otros lenguajes.

import abc

class Animal:

    @abc.abstractmethod
    def sonido(self):
        pass

class Perro(Animal):

    def sonido(self):
        print("Ladrido")

class Vaca(Animal):

    def sonido(self):
        print("Mugido")

Se puede utilizar en conjunción con @property o staticmethod para crear propiedades o métodos estáticos abstractos.

functools.total_ordering

Nos permite crear clases que soporten operadores de comparación. Solo tenemos que proveer el método _ eq _ y uno de los seis métodos predefinidos de orden. La documentación de la bilioteca estándar advierte que este decorador puede tener impactos en el rendimiento.

@total_ordering
class Student:
    def _is_valid_operand(self, other):
        return (hasattr(other, "lastname") and
                hasattr(other, "firstname"))
    def __eq__ (self, other):
        if not self._is_valid_operand(other):
            return NotImplemented
        return ((self.lastname.lower(), self.firstname.lower()) ==
                (other.lastname.lower(), other.firstname.lower()))
    def __lt__ (self, other):
        if not self._is_valid_operand(other):
            return NotImplemented
        return ((self.lastname.lower(), self.firstname.lower()) <
                (other.lastname.lower(), other.firstname.lower()))

dataclasses.dataclass

Nos permite omitir la declaración de _ init _ y _ repr _ en la declaración de clases en las que solo vamos a manejar datos. El uso puede ser extendido a subclases

import dataclasses

@dataclasses.dataclass
class Point:
    x: float
    y: float

    def euclidian_distance(self, other: Point) -> float:
        return math.sqrt(
            (other.x - self.y)**2 + (other.y - self.y)**2
        )

@dataclasses.dataclass
class ThreeDPoint(Point):
    z: float

    def euclidian_distance(self,other: Point) -> float:
        dx = (other.x - self.y)**2
        dy = (other.y - self.y)**2
        dz = (other.z - self.z)**2
        return math.sqrt(
            dx+dy+dz
        )

Decoradores para funciones

functools.lru_cache

Caché de las últimas n llamadas a una función. Los parametros tienen que poder ser utilizados como llaves de un diccionario.

from functools import lru_cache
@lru_cache(20)
def get_exchange(local, foreign):
    exchange_service.convert(local, foreign)

asyncio.coroutine

Convierte corutinas basadas en generadores para que sean compatibles con el nuevo modelo async/await. Usar solamente en código que necesita ser portado

from asyncio import coroutine

@coroutine
def render_view(view_name, context):
    template = Template(view_name, context)
    yield template.render()

async def hanlde_get(route):
    await render_view(route.view, route.params)

contextlib.contextmanager

Convierte una función en un contextmanager. Puede ser usado de dos formas:

Utilizando una sentencia with
Como un decorador.

Personalmente para hacer decoradores que sean contextmanagers prefiero utilizar la clase ContextDecorator.

from contextlib import contextmanager

@contextmanager
def acquire_lock(key):
    credentials = get_credentials(key)
    lock = acquire_lock_using_credentials(credentials)
    try:
        yield lock
    finally:
        release_lock(lock)

with acquire_lock('my secret api key') as lock:
    use_critical_resource(lock)

Decoradores extra

functools.wraps

Decorador para crear decoradores. Útil para hacer que las funciones conserven propiedades como _ name _ y _ doc _

from functools import wraps
def my_decorator(f):
    @wraps(f)
    def wrapper(*args, **kwds):
        return f(*args, **kwds)
        return wrapper

@my_decorator
def funcion():
    """
    Esta es la doc
    """
    pass

functools.singledispatch

Permite crear funciones sobrecargadas a partir del tipo del primer parámetro. Hace uso de las anotaciones de tipos.

class Perro:
    pass

clas Vaca:
    pass

@singledispatch
def sonido(animal):
    pass

@sonido.register(Perro)
def _(animal):
    print("Ladra")

@sonido.register(Vaca)
def _(animal):
    print("Muge")

No soy muy partidario de usar métodos estáticos en Python. ^[return]