Skip to content
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):
Want to print your doc?
This is not the way.
This is not the way.
Try clicking the ⋯ next to your doc name or using a keyboard shortcut (
CtrlP
) instead.