Explore

Realtime Events

import asyncio

from supabase import acreate_client, AClient

url: str = "my-soko-api-url"

key: str = "my-soko-api-key"

async def setup_database_realtime():

supabase: AClient = await acreate_client(url, key)

def on_subscribe(status, err):

print(status)

user = await supabase.auth.sign_in_with_password({

"email": "user@sokovfx.com",

"password": "mypassword"

})

def handle_record_updated(payload):

print(payload["data"]["table"])

await supabase.realtime.connect()

await supabase.realtime.channel('postgres_changes').on_postgres_changes("*", schema="*",

callback=handle_record_updated).subscribe(

on_subscribe)

await supabase.realtime.listen()

asyncio.run(setup_database_realtime())

Actions

Main repository is Pipeline-Deadline-Integration

all action scripts are palced uder plugins/Soko/Jobs folder

In Soko Electron App we use emdeded Python 3.11.7 so it’s not necessary to install it on PC separate, but it is required to install it when you vant to develop your own actions

https://www.python.org/downloads/release/python-3117/

Every Action inherits from the class SokoJob which looks like:

import json

from soko_api.client import create_client, Client

import logging

import sys

# Configure logging to output to stdout

logging.basicConfig(

level=logging.INFO,

format='%(levelname)s: %(message)s',

stream=sys.stdout # Redirect logs to stdout

)

class SokoJob:

def __init__(self, args):

# Get Job arguments from aruments file

params_file = args[-1]

self.arguments = {}

with open(params_file) as f:

self.arguments = json.load(f)

self.version = "unknown"

self.label = None

self.group = None

self.icon = None

self.isForProject = False

self.isForFolder = False

self.isForVersion = False

self.isForMultipleObjects = False

self.askForConfirmation = False

self.suppressAskForFilesSelect = False

self.runLocal = False

self.widget = None

self.failureDetectionJobErrors = None

def Handle(self):

# Override this method with Job logic

...

def InitApi(self):

# Init Soko API

url: str = self.arguments["pluginData"]["serverURL"]

key: str = self.arguments["pluginData"]["apiKey"]

api_user = self.arguments["pluginData"].get("apiUser")

api_user_password = self.arguments["pluginData"].get("apiUserPassword")

jwt = self.arguments["pluginData"].get("jwt")

user_api_key = self.arguments["pluginData"].get("userApiKey")

if api_user:

self.soko_api: Client = create_client(url, key)

self.soko_api.sign_in_with_password(api_user, api_user_password)

elif user_api_key:

self.soko_api: Client = create_client(url, key)

self.soko_api.sign_in_with_api_key(user_api_key)

elif jwt:

self.soko_api: Client = create_client(url, key)

self.soko_api.sign_in_with_jwt(jwt)

else:

raise Exception("Failed to login into Soko API.")

Where Handle is main body function of the action

Creating the custom action

import logging

import os

import sys

import traceback

import uuid

from soko_api.client import create_client, Client

from soko_api.utils import video

from soko_api.utils import files

import subprocess

from SokoJob import SokoJob

# Inherit fom SokoJob class

class MyAction(SokoJob):

def __init__(self, args):

super().__init__(args)

self.version = "1.0.0"

self.icon = "video"

self.isForVersion = True

self.label = "My New Action"

self.group = "Utils"

self.runLocal = True

self.failureDetectionJobErrors = 3

def Handle(self):

# Overide Handle method with custom Job logic

logging.warning("MyAction")

self.InitApi()

file_types_res = self.soko_api.table("fileTypes").select("*").execute()

sokoFileTypesMap = {}

if file_types_res is not None and file_types_res.data:

sokoFileTypesMap = {

k["fileExtension"]: k["id"] for k in file_types_res.data

}

for contextObject in self.arguments["contextObjects"]:

for component in contextObject["components"]:

_component_path_data = files.get_component_path_data_for_current_os(component['parsedPaths'])

_component_path = _component_path_data["path"]

# do something usefull

if __name__ == "__main__":

# Init Job object

plugin = MyAction(sys.argv)

# Run job

plugin.Handle()

When user runs an action from Soko, application basicaly do:

1. Action script is beighn downloaded from the server and is saved as temporary file.

For example on Windows:

C:/Users/blue/AppData/Local/Temp/tmp-31324-o3tW5RyOI6Mf/script.py

2. The Action arguments are passed from Soko App as object and they are also saved into the temporary location as json file.

For example on Windows:

C:/Users/blue/AppData/Local/Temp/tmp-31324-o3tW5RyOI6Mf/arguments.json

Than Soko App runs the action as python script:

C:\soko\python\python.exe C:/Temp/tmp-31324-o3tW5RyOI6Mf/script.py C:/Temp/tmp-31324-o3tW5RyOI6Mf/arguments.json

Every action script starts here:

if __name__ == "__main__":

# Init Job object

plugin = MyAction(sys.argv)

# Run job

plugin.Handle()

Where sys.argv are standard command line arguments and in python are everytime in order:

1. is always the script file itself -> script.py

2..N are all another arguments

As yo can see, the instance of your action is created each time:

plugin = MyAction(sys.argv)

which call super().__init__(args) in the __init__ method:

class MyAction(SokoJob):

def __init__(self, args):

super().__init__(args) # <------- calls SokoJob.__init__

Which loads arguments file (we expect the arguments.json is always the last argument):

class SokoJob:

def __init__(self, args):

# Get Job arguments from aruments file

params_file = args[-1]

self.arguments = {}

with open(params_file) as f:

self.arguments = json.load(f) # <--------- save argument data for later on

Based on context object type (object on you run action), arguments will look like:

⁠

Context object is always represented as the list of objects in contextObjects property. Action can be marked to be used for:

isForProject - Project Action

isForFolder - Folder Action

isForVersion - Version Action

Than plugin.Handle() method is called

Uploading Action into Soko Cloud

When you have your own action, you can simply upload it into the Soko:

Navigate to Settings/Pipeline/Actions

Press the “+” button

Drag & Drop or browse your script file

Press “Upload”

⁠

Now you can find your action:

⁠

Want to print your doc?
This is not the way.

Try clicking the ··· in the right corner or using a keyboard shortcut (

CtrlP

) instead.