pyport.exs

defmodule Pyport do
  @pool_size 10
  @external_resource "pyport.py"
  use GenServer

  def start_link(args, opts \\ [name: __MODULE__]) do
    GenServer.start_link(__MODULE__, args, opts)
  end

  def echo(pid \\ __MODULE__, data) do
    GenServer.call(pid, {:work, {:echo, data}})
  end

  def upload_image(pid \\ __MODULE__, content, filename, content_type, timeout \\ 5000) do
    GenServer.call(pid, {:work, {:upload_image, content, filename, content_type}}, timeout)
  end

  def init(_args) do
    send(self(), :init_pool)
    {:ok, %{pool: [], index: 0}}
  end

  def handle_info(:init_pool, state) do
    python = System.find_executable("python3")
    pool = Enum.map(0..@pool_size, fn(_) -> 
      :erlang.open_port({:spawn_executable, python}, [
        :binary,
        {:packet, 4},
        # {:args, ["-c", @python_source]},
        {:args, ["-u", "pyport.py"]},
        :nouse_stdio
      ])
    end)

    {:noreply, %{state | pool: pool}}
  end

  def handle_info({port, {:data, data}}, state) when is_port(port) do
    {result, from} = :erlang.binary_to_term(data)
    GenServer.reply(from, result)
    {:noreply, state}
  end

  def handle_call({:work, command}, from, state) do
    {:noreply, state, {:continue, {command, from}}}
  end

  def handle_continue({command, from}, state) do
    port = Enum.at(state.pool, state.index)
    true = :erlang.port_command(port, :erlang.term_to_binary({command, from}))
    if state.index == @pool_size do
      {:noreply, %{state | index: 0}}
    else
      {:noreply, %{state | index: state.index + 1}}
    end
  end
end

pyport.py

#!/usr/bin/env python
import erlang, os, struct
from google.cloud import storage
import six

bucketname = "TOP SECRET"
# [START bookshelf_cloud_storage_client]
client = storage.Client()
bucket = client.bucket(bucketname)

def upload_image(image_binary_data, filename, content_type):
  blob = bucket.blob(filename)
  blob.upload_from_string(
      image_binary_data,
      content_type=content_type
  )

  url = blob.public_url

  if isinstance(url, six.binary_type):
    url = url.decode('utf-8')
  
  return url

def send(term, stream):
    payload = erlang.term_to_binary(term)
    header = struct.pack('!I', len(payload))
    stream.write(header)
    stream.write(payload)
    stream.flush()

def recv(stream):
    header = stream.read(4)
    if len(header) != 4:
        return None # EOF
    (length,) = struct.unpack('!I', header)
    payload = stream.read(length)
    if len(payload) != length:
        return None
    term = erlang.binary_to_term(payload)
    return term

def recv_loop(stream):
    message = recv(stream)
    while message:
        yield message
        message = recv(stream)

AtomUpload = erlang.OtpErlangAtom(bytes("upload_image", "utf-8"))

if __name__ == '__main__':
    input, output = os.fdopen(3, 'rb'), os.fdopen(4, 'wb')
    for message in recv_loop(input):
        command = message[0]
        operation = command[0]
        otp_from = message[1]
        if operation == AtomUpload:
          result = upload_image(command[1].value, command[2].value, command[3].value)
          send((result, otp_from), output)
        else:
          send(message, output) # echo the message back