Classes

Thread workers

class pantra.workers.base.BaseWorkerServer

Abstract thread management and communication server class

It has basic thread manager implementation and abstract communication methods. Read Threading model and Message queue for more information

class Listener

Abstract communication listener class

abstractmethod async send(session_id: str, data: bytes)

Send packed message to client by session_id

Parameters:

• session_id (str)

• data (bytes)

abstractmethod async receive() → tuple[str, bytes]

Read packed message from client and return tuple (session_id, data)

Return type:

tuple[str, bytes]

abstractmethod async close()

Close active connection

classmethod task_processor()

Single task thread processor.

It extracts new function call with args from task_queue and let it run in this thread

classmethod tasks_controller()

Common task controller

It supervises all task threads life cycle. See also Threading model.

static session_killer()

Session TTL controller

abstractmethod start_listener()

Main entrypoint for listener subclass initialization

it should end by:

self.listener = WorkerServer.Listener(...)

class pantra.workers.base.BaseWorkerClient(session_id: str, ws: WebSocket, app: str, lang: list[str], params: dict[str, str])

Abstract thread management and communication client class

It contains abstract communication methods.

class Connection

Abstract communication connection class

abstractmethod async send(message: bytes)

Send a packed message to the server

Parameters:

message (bytes)

abstractmethod async receive() → bytes

Receive a packed message from the server

Return type:

bytes

abstractmethod close()

Close the connection

abstractmethod open_connection(session_id: str)

Main entrypoint for connection subclass initialization

it should end by:

self.connection = WorkerClient.Connection(...)

Parameters:

session_id (str)

Routers

class pantra.routes.BaseRouter

Basic router class.

It has all needed to startup and shutdown router, and also has websocket processor. All other derived classes are responsible for processing: * static files, including engine JS * any additional API calls

It is recommended to inherit from DevRouter or CachedRouter instead of base class

static_routes() → list[Mount]

Define static routes

Return type:

list[Mount]

static forbidden(message: str) → Response

Shortcut method to return 403

Parameters:

message (str)

Return type:

Response

static not_found(message: str) → Response

Shortcut method to return 404

Parameters:

message (str)

Return type:

Response

static bad_request(message: str = '') → Response

Shortcut method to return 400

Parameters:

message (str)

Return type:

Response

static css(value: str) → Response

Shortcut method to return CSS content

Parameters:

value (str)

Return type:

Response

class pantra.routes.DevRouter

Router intended for development process.

It provides logic to use raw components, generate data on the fly, slow yet handy

class pantra.routes.CachedRouter

Router intended for production.

It uses pre-cached components data only. Fast and efficient.

pantra.routes.route(pattern: str, method: str = None)

The decorator to mark method as a router to specific regex pattern

Parameters:

• pattern (str)

• method (str)

pantra.routes.get(pattern: str)

Decorator shortcut for GET method

Parameters:

pattern (str)

pantra.routes.post(pattern: str)

Decorator shortcut for POST method

Parameters:

pattern (str)

Drag’n’drop action

class pantra.components.controllers.DragOptions(mouse_button: int = 0, allow_move_by_x: bool = True, allow_move_by_y: bool = True, border_top: int | None = None, border_left: int | None = None, border_bottom: int | None = None, border_right: int | None = None)

Initial drag’n’drop control options

Parameters:

• mouse_button (int)

• allow_move_by_x (bool)

• allow_move_by_y (bool)

• border_top (Optional[int])

• border_left (Optional[int])

• border_bottom (Optional[int])

• border_right (Optional[int])

mouse_button

code of mouse button to push for drag

Type:

int

allow_move_by_x

allow move dragged box by x-axis

Type:

bool

allow_move_by_y

allow move dragged box by y-axis

Type:

bool

border_top

drag zone border top max coordinate

Type:

Optional[int]

border_bottom

border bottom max coordinate

Type:

Optional[int]

border_left

border left max coordinate

Type:

Optional[int]

border_right

border right max coordinate

Type:

Optional[int]

class pantra.components.controllers.DragController(node: HTMLElement)

Abstract class to perform drag’n’drop action

Parameters:

node (HTMLElement)

from_x

initial x coordinate

from_y

initial y coordinate

x

current x coordinate

y

current y coordinate

delta_x

changes of x coordinate since last time

delta_y

changes of y coordinate since last time

options

control options

in_moving

box in moving process

get_options(node: HTMLElement)

get options for node

Override this method for custom options for specified node

Parameters:

node (HTMLElement)

abstractmethod start(node) → bool

method invoked when drag action performed and satisfied to options

Return type:

bool

abstractmethod move()

method invoked on drag box moving

abstractmethod stop()

method invoked on drag box drop

Common classes

class pantra.common.HTML

Mark string as safe HTML content.

Example:

refs["message"].set_text(HTML("<h1>Big one</h1>"))

class pantra.common.DynamicDict(*args, _lazy_mode: bool = False, **kwargs)

Dictionary with dynamic (computable) values.

It allows to set lambda functions for evaluation and repeated evaluations later:

x = 2
d = DynamicDict()
d["f"] = lambda : x**2
print(d["f"]) # -> 4
x = 3
d.refresh()
print(d["f"]) # -> 9

Parameters:

_lazy_mode (bool) – don’t evaluate functions at the first assignment

items()

Return a set-like object providing a view on the dict’s items.

refresh(attr_name: str | None = None)

repeated evaluations

Affects all functions if attr_name is not specified.

Parameters:

attr_name (str) – attribute name for repeated evaluations.

refresh_items() → Iterable[tuple[str, Any]]

iterate through evaluated items only

Return type:

Iterable[tuple[str, Any]]

class pantra.common.DynamicClasses(*args, **kwargs)

Extension to DynamicDict to support HTML classes manipulation.

It allows basic arithmetic operations intuitively:

c = DynamicClasses("colored bold animated")
c['selected'] = lambda: True
c += "hidden"
c -= "bold"
c *= "active"
c *= "active" # switch class on and off again
print(c) # -> "colored animated selected hidden"

class pantra.common.DynamicStyles(style: dict | str | None = None)

Extension to DynamicDict to support styles parsing

Example:

s = DynamicStyles('left: 0; top: 0; width: 100%; height: 100%')
print(s['left']) # -> "0"
print(s['width']) # -> "100%"
s -= "width"
s["position"] = lambda: "absolute"
s += "width: 80%"
print(s) # -> "left: 0; top: 0; height: 100%; width: 80%; position: absolute"

Parameters:

style (dict | str | None)

class pantra.common.WebUnits(value: int | float | str, unit: str = 'px')

Helper type to work with web units

This time supports natural arithmetic operations: + - * / //

Parameters:

• value (int | float | str) – numeric value

• unit (str) – unit name (px | em | % | …)

Session class

class pantra.session.Session(session_id: str, app: str, lang: list[str], params: dict[str, str])

Session class.

Session is created on new user connection and contains isolated user-related data.

Parameters:

• session_id (str)

• app (str)

• lang (list[str])

• params (dict[str, str])

session_id

unique session id, generated on client side

app

current application name

Type:

str

title

application title to the browser page (see set_title())

user

current user ID, if used

state

any user-related data, associated with active session

Type:

dict[str, Any]

root

root component (always “Main”)

Type:

Context

locale

current locale (see set_locale() and more)

Type:

Locale

storage

permanent settings storage (see more)

Type:

SessionStorage

params

URL params (http://localhost/app/?a=1&b=2&c=3)

Type:

dict[str, str]

last_touch

last time event was triggered on this session

Type:

datetime

tasks

all tasks running (see more)

Type:

dict[str, SessionTask]

sessions

(class variable) all sessions collection

Type:

dict[str, Session]

pending_errors

(class variable) all pending errors queue, to send to next user on next session

Type:

Queue[str]

server_worker

(class variable) main server worker to host all sessions

Type:

BaseWorkerServer

static gen_session_id() → str

simple unique session id generator

Return type:

str

get_node(oid: int) → AnyNode | None

shortcut to get node by OID (read here)

Parameters:

oid (int)

Return type:

AnyNode | None

restart()

restart application and rebuild main context

error(text: str, exc_value: Exception | None = None)

send error message to client-side

Parameters:

• text (str) – error text

• exc_value (Exception | None) – exception value with reference to node

static error_later(message)

collect pending errors outside the active session

send_shot()

send DOM changes snapshot

kill_task(task_name: str)

kill task by name

Parameters:

task_name (str)

kill_all_tasks(ctx: UniqueNode = None)

kill all tasks for specified context and all

Parameters:

ctx (UniqueNode)

log(text: str)

send message to user browser’s debug console

Parameters:

text (str)

call(method: str, *args)

call JavaScript method by name and arguments

Parameters:

method (str)

static get_apps() → list[str]

get list of available apps

Return type:

list[str]

start_app(app: str)

send message to switch to and start app

Parameters:

app (str)

set_title(title: str)

set title and send to client-side

Parameters:

title (str)

send_task_done()

send signal task finished, to remove waiting (spinner) animation

set_locale(lang: str | list)

set locale from one or several languages specified

Parameters:

lang (str | list)

zgettext(message: str, *args, plural: str = None, n: int = None, ctx: str = None, many: bool = False)

unite translation function

Parameters:

message (str)

Return type:

str

gettext(message: str) → str

simple translation function

Parameters:

message (str)

Return type:

str

set_storage(storage_cls: type[SessionStorage] | SessionStorage)

set session storage

Parameters:

storage_cls (type[SessionStorage] | SessionStorage)

bind_state(name: str, dict_ref: dict[str, Any], key: str = 'value')

bind component state to session storage

Parameters:

• name (str) – key name in storage

• dict_ref (dict[str, Any]) – binding mapping

• key (str) – key name in binding mapping

bind_states(bindings: dict[str, dict | tuple[dict, str]])

bind several components states using short form of dict

Parameters:

bindings (dict[str, dict | tuple[dict, str]]) – dict of dict references or tuples of dict references and keys

Example:

session.bind_states({
    "api_token": refs['token'],
    "api_user_id": refs['user_id'],
    "other_value": (refs['other'], "property")
})

sync_storage()

gather changes and save to the session storage

key_events_off()

send signal to disable all key events

key_events_on()

send signal to enable all key events