Packages
snakepit
0.8.4
0.13.0
0.12.0
0.11.1
0.11.0
0.10.1
0.10.0
0.9.1
0.9.0
0.8.9
0.8.8
0.8.7
0.8.6
0.8.5
0.8.4
0.8.3
0.8.2
0.8.1
0.8.0
0.7.7
0.7.6
0.7.5
0.7.4
0.7.3
0.7.2
0.7.1
0.7.0
0.6.11
0.6.10
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.2
0.6.1
0.6.0
0.5.1
0.5.0
0.4.3
0.4.2
0.4.1
0.4.0
0.3.3
0.3.2
0.3.1
0.3.0
0.2.1
0.2.0
0.1.2
0.1.1
0.1.0
High-performance pooler and session manager for external language integrations. Supports Python, Node.js, Ruby, and more with gRPC streaming, session management, and production-ready process cleanup.
Current section
Files
Jump to
Current section
Files
lib/snakepit/error/python_translation.ex
defmodule Snakepit.Error.PythonTranslation do
@moduledoc false
alias Snakepit.Error
@mapping %{
"ValueError" => Error.ValueError,
"KeyError" => Error.KeyError,
"IndexError" => Error.IndexError,
"TypeError" => Error.TypeError,
"AttributeError" => Error.AttributeError,
"ImportError" => Error.ImportError,
"RuntimeError" => Error.RuntimeError,
"NotImplementedError" => Error.NotImplementedError,
"FileNotFoundError" => Error.FileNotFoundError,
"PermissionError" => Error.PermissionError,
"ZeroDivisionError" => Error.ZeroDivisionError
}
def from_error_message(error_message) when is_binary(error_message) do
with {:ok, payload} <- Jason.decode(error_message),
true <- is_map(payload),
{:ok, exception} <- from_payload(payload) do
{:ok, exception}
else
_ -> :error
end
end
def from_error_message(_), do: :error
def from_payload(payload) when is_map(payload) do
type = fetch(payload, ["type", :type])
message = fetch(payload, ["message", :message])
if is_binary(type) and is_binary(message) do
context = fetch(payload, ["context", :context]) || %{}
stacktrace = fetch(payload, ["stacktrace", :stacktrace]) || []
python_traceback = fetch(payload, ["traceback", :traceback]) || join_stacktrace(stacktrace)
module = Map.get(@mapping, type, Error.PythonException)
mapped? = Map.has_key?(@mapping, type)
exception =
struct(module, %{
message: message,
context: context,
stacktrace: stacktrace,
python_type: type,
python_traceback: python_traceback
})
emit_exception_event(mapped?, type, context)
{:ok, exception}
else
:error
end
end
def from_payload(_), do: :error
defp fetch(payload, keys) do
Enum.find_value(keys, fn key -> Map.get(payload, key) end)
end
defp join_stacktrace(stacktrace) when is_list(stacktrace) do
Enum.map_join(stacktrace, "", &to_string/1)
end
defp join_stacktrace(_), do: nil
defp emit_exception_event(mapped?, python_type, context) do
event =
if mapped? do
[:snakepit, :python, :exception, :mapped]
else
[:snakepit, :python, :exception, :unmapped]
end
metadata = %{
python_type: python_type,
library: fetch(context, ["library", :library]),
function: fetch(context, ["function", :function])
}
:telemetry.execute(event, %{}, metadata)
end
end