pythonrt

README

Python Runtime

Implementation of Python runtime. See Python runtime for list of issues.

Currently, we don't support 3rd party packages (e.g. pip install) for the user code. See ENG-538 for more details. For a realistic POC/demo, we'll pre-install the packages the user code requires (e.g. slack-sdk).

Python Handler Function

Python's handler functions should receive a single dict value which is the triggering event. This event has the following keys:

• event_type (str): The event type
• event_id (str): The event ID
• integration_id (str): The ID of the integration that fired the event
• data (dict): The event payload (integration specific)

The return value from Python entry points is ignored and is not passed back to autokitteh.

Limitations

You can't issue function calls at module level (e.g. TOKEN = os.getenv('TOKEN'))

Python serializes function calls using pickle, some callables can't be pickled:

• lambda
• dynamically generate functions (notably os.environ.get)

Patching User Code

The Python code (ak_runner.py) loads the user code and patches every function call. It does so by hooking into the regular import hook. When the user module is loaded, we transform the AST to change function calls from:

urlopen(url)

to:

_ak_call(urlopen, url)

ak_call will call the Go Python runtime which will start an activity.

Detecting External Function Calls

When the user module is loaded, we set MODULE_NAME to the name of the loaded module. When ak_call is invoked with a function it compares the function module name with MODULE_NAME, if they are the same, it's an internal function and ak_call will return the invocation of the function. Otherwise, ak_call will call the Go process that will invoke an activity.

See ENG-495.

Go ↔ Python Communication Flow

A run calls start a Python server with:

• Tar file containing user code
• Entry point (e.g. review.py:on_github_pull_request)

It will also inject vars definition from the manifest to the Python process environment.

The Python server returns a list of exported symbols from the user code.

Communication Sequence

A call with function and payload:

sequenceDiagram
    Python-->>Go: Module loaded (exports)
    Go->>Python: Run(function, payload)
    loop
        Python-->>Go: Activity request (function name, args, payload)
        Go ->>Python: Activity call (payload, skipped in reply)
        Python-->>Go: Activity value (skipped in reply)
        Go ->>Python: Activity Value
    end
    Python-->>Go: Run return (None)

Other messages are:

• log from Python to Go
• sleep from Python to Go

State Machine

Below is the state machine for Call:

---
title: Call State Machine
---
stateDiagram-v2
    Wait --> Connected: exports
    Connected --> Running : run
    Running --> Done : done
    Running --> Error : empty
    Running --> Running : log
    Running --> Running: func
    Running --> CallFromPython : call
    CallFromPython --> CallToPython : return
    CallFromPython --> ReturnToPython: replay
    CallToPython --> CallToPython: log
    CallToPython --> CallToPython: func
    CallToPython --> Error: empty
    CallToPython --> ReturnFromPython: return
    ReturnFromPython --> ReturnToPython: return
    ReturnToPython  --> Running
    ReturnToPython --> Error: empty

Communication Protocol

We're using JSON over Unix domain socket, one JSON object per line. The reason do this is that ak_runner.py should not have any external dependencies outside the standard library. Once we introduce an external dependency, it will conflict with the user dependencies.

All messages have top level type and payload, the payload changes depending on the type.

run

• func_name: Function name (string)
• event: Data payload (map)

module

• entries: List of exported functions (array of strings)

callback

• name: Function name (string)
• args: Function args (array of strings)
• kw: Mapping of name → value (both strings)
• data: Pickled (protocol 0) tuple of (func, args, kw) encoded in base64

response

• value: Return type from Python (base64 of Python's pickle protocol 0)

done

• No fields

Integration Testing

To make sure you use autokitteh from the py-sdk, update your PYTHONPATH before running ak. Run the following from the root of the repo:

export PYTHONPATH="${PWD}/runtimes/pythonrt/py-sdk:${PYTHONPATH}"

If you run ak with a database, then run make create-workflow once. Otherwise run it every time. This will create a deployment for testdata/simple/

Then run make run-workflow.

ak with database

Look for the config.yaml in ak config where directory. Then add the following

db:
  dsn: /tmp/ak.db  # Pick any other location
  type: sqlite

Hacking

The ak_runner is directory in ak, you'll need to build it (make bin from root of project) every time you change Python code and want to run workflows.

Your code should be formatted with ruff.

Documentation

Index

• Variables
• func New() (sdkservices.Runtime, error)
• type CallMessage
  • func (CallMessage) Type() string
• type CallbackMessage
  • func (CallbackMessage) Type() string
• type Comm
  • func NewComm(conn net.Conn) *Comm
  • func (c *Comm) Close() error
  • func (c *Comm) Recv() (Message, error)
  • func (c *Comm) Send(msg Typed) error
• type DoneMessage
  • func (DoneMessage) Type() string
• type ErrorMessage
  • func (ErrorMessage) Type() string
• type Export
• type Frame
• type LogMessage
  • func (LogMessage) Type() string
• type Message
• type ModuleMessage
  • func (ModuleMessage) Type() string
• type ResponseMessage
  • func (ResponseMessage) Type() string
• type ReturnMessage
  • func (ReturnMessage) Type() string
• type RunMessage
  • func (RunMessage) Type() string
• type SubMessage
• type Typed
• type Version

Variables

var (
	Runtime = &sdkruntimes.Runtime{
		Desc: kittehs.Must1(sdktypes.StrictRuntimeFromProto(&sdktypes.RuntimePB{
			Name:           "python",
			FileExtensions: []string{"py"},
		})),
		New: New,
	}
)

Functions

func New

func New() (sdkservices.Runtime, error)

Types

type CallMessage

type CallMessage struct {
	FuncName string         `json:"func_name"`
	Args     []any          `json:"args"`
	Kw       map[string]any `json:"kw"`
}

func (CallMessage) Type

func (CallMessage) Type() string

type CallbackMessage

type CallbackMessage struct {
	Name string            `json:"name"`
	Args []string          `json:"args"`
	Kw   map[string]string `json:"kw"`
	Data []byte            `json:"data"`
}

py <-> go

func (CallbackMessage) Type

func (CallbackMessage) Type() string

type Comm

type Comm struct {
	// contains filtered or unexported fields
}

func NewComm

func NewComm(conn net.Conn) *Comm

func (*Comm) Close

func (c *Comm) Close() error

func (*Comm) Recv

func (c *Comm) Recv() (Message, error)

func (*Comm) Send

func (c *Comm) Send(msg Typed) error

type DoneMessage

type DoneMessage struct{}

func (DoneMessage) Type

func (DoneMessage) Type() string

type ErrorMessage

type ErrorMessage struct {
	Error     string  `json:"error"`
	Traceback []Frame `json:"traceback"`
}

func (ErrorMessage) Type

func (ErrorMessage) Type() string

type Export

type Export struct {
	Name string
	File string
	Line int
}

type Frame

type Frame struct {
	File   string `json:"file"`
	LineNo uint32 `json:"lineno"`
	Code   string `json:"code"`
	Name   string `json:"name"`
}

type LogMessage

type LogMessage struct {
	Level   string `json:"level"`
	Message string `json:"message"`
}

func (LogMessage) Type

func (LogMessage) Type() string

type Message

type Message struct {
	Type    string          `json:"type"`
	Payload json.RawMessage `json:"payload"`
}

type ModuleMessage

type ModuleMessage struct {
	Entries []string `json:"entries"`
}

py -> go

func (ModuleMessage) Type

func (ModuleMessage) Type() string

type ResponseMessage

type ResponseMessage struct {
	Value []byte `json:"value"`
}

py <-> go

func (ResponseMessage) Type

func (ResponseMessage) Type() string

type ReturnMessage

type ReturnMessage struct {
	Value any `json:"value"`
}

func (ReturnMessage) Type

func (ReturnMessage) Type() string

type RunMessage

type RunMessage struct {
	FuncName string         `json:"func_name"`
	Event    map[string]any `json:"event"`
}

go -> python

func (RunMessage) Type

func (RunMessage) Type() string

type SubMessage

type SubMessage interface {
	CallMessage |
		CallbackMessage |
		ErrorMessage |
		DoneMessage |
		LogMessage |
		ModuleMessage |
		ResponseMessage |
		ReturnMessage |
		RunMessage

	Typed
}

type Typed

type Typed interface {
	Type() string
}

type Version

type Version struct {
	Major int
	Minor int
}