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()
Want to print your doc?
This is not the way.
This is not the way.
Try clicking the ··· in the right corner or using a keyboard shortcut (
CtrlP
) instead.