LiveView Python Client Documentation

The LiveView Python client lets you query, publish, and monitor data from a LiveView server with Python.

Getting Started

Python version >= 3.6 is required.

To install the LiveView Python client:

# Make sure $STREAMBASE_HOME is set
$(eval sb-config --env --for-command-prompt)

pip install $STREAMBASE_HOME/sdk/python/dist/lv_python_client-11.0.0-py3-none-any.whl

Run the following in a Python interpreter to verify that the Python client installed correctly:

from tibco import liveview

client = liveview.get_client('lv://localhost:11080')

# If you have a LiveView server running on port 11080, this should succeed
client.ensure_connection()

Snapshot and Live Queries

The client can run snapshot and live queries:

from tibco import liveview
import time
from tibco.liveview.listeners import QueryListener
from tibco.liveview import build_query as q

# Get a client for a server running Hello LiveView
client = liveview.get_client('lv://localhost:11080')

# Perform a snapshot query
query = q('ItemsInventory').select('*')
snapshot = client.snapshot_query(query).to_dict()

# Inspect the snapshot's schema and tuples
print(snapshot['schema'])
print(snapshot['data'])

# Prepare a live query
live_query = client.live_query(q('ItemsSales').select('*'))
# Or
# live_query = client.live_query_from_s('ItemsSales', 'SELECT * FROM ItemsSales')

# Define a callback function which will be called with tuples whenever a tuple add event is received
def tuple_added(lv_tuple):
    print(lv_tuple)

with QueryListener(live_query, tuple_added=tuple_added):
    # Let the query listener run for 3 seconds
    time.sleep(3)

# Now the query listener has stopped listening.

Live Result

A LiveResult listens to a live query and maintains a result set of tuples as add, remove, and update events come in.

WARNING: A LiveResult is intended for use only with queries that do not infinitely grow. A LiveResult does not have a maximum capacity. Limit clauses may be used to guarantee one.

import time
from tibco import liveview
from tibco.liveview.listeners import LiveResult

client = liveview.get_client('lv://localhost:11080')
live_query = client.live_query_from_s('ItemsSales', 'SELECT * FROM ItemsSales')

live_result = LiveResult(live_query,
                         exception_raised=lambda e: print(e),
                         # Convert incoming tuples into Python types using the query's schema
                         convert_with_schema=True)

# Get some results
with live_result:
    time.sleep(3)

'''The above is equivalent to
live_result.start()
time.sleep(3)
live_result.stop()
'''

# Inspect the results.
items, categories = live_result.get_columns('Item', 'category')
for item, category in zip(items, categories):
    print(f'Item: {item}, Category: {category}')

Query Builder

The LiveView Python Client comes with a query builder that allows you to intuitively construct queries:

from tibco import liveview
from tibco.liveview.listeners import LiveResult
from tibco.liveview import build_query as q

query = q('ItemsSales').select('Item', 'category').where(category='electronics')
client = liveview.get_client('lv://localhost::11080')

live_query = client.live_query(query)

API Reference

LiveViewClient

class LiveViewClient(uri: str, username: str = '', password: str = '')

Provides interface for all LiveView Python client functionality.

Parameters

• uri – The URI string for the client to connect to, for example: 'lv://localhost:11080'

• username (optional) – LiveView server username

• password (optional) – LiveView server password

delete_tuples(table_name: str, tuples: List) → None

Deletes tuples given a table name and a list of tuples.

Parameters

• table_name (str) – The table from which to delete tuples.

• tuples (list) – A list of tuple primary keys which point to tuples to delete

delete_tuples_by_query(query: tibco.liveview.lv_query.Query)

Deletes tuples retrieved by a query.

Parameters

query (Query) – A query whose results should be deleted.

delete_tuples_by_query_s(table_name: str, query_string: str)

Deletes tuples given a table name and a query string.

Parameters

• table_name (str) – A table from which to delete tuples.

• query_string (str) – A query over table_name whose results should be deleted.

ensure_connection() → None

Raises a LiveViewException if not connected.

Raises

LiveViewException – if not connected

get_buffered_publisher(table_name: str, publish_interval: float = 1) → tibco.liveview.models.buffered_publisher.BufferedLVPublisher

Parameters

• table_name – name of table for publishing to

• publish_interval – every publish_interval seconds, the publisher will publish any tuples in its buffer

Returns

A publisher which maintains a buffer that it will try to flush to an LV server at a specified interval.

get_table_metadata(table_name, include_internal=False) → tibco.liveview.models.table_metadata.TableMetadata

Get metadata for a table

Parameters

• table_name (str) – The name of the table.

• include_internal (bool) – Whether to include internal LiveView fields

Returns

A TableInfo object containing the tables metadata.

list_tables() → list

Lists the names of every table on the server.

Returns

• A list of names (strings)

live_query(query: tibco.liveview.lv_query.Query)

Prepare a live query from a Query object.

Usage:

import liveview
from liveview import Query

client = liveview.get_client('lv://localhost:11080')
live_query = client.live_query(Query('ItemsSales').select('*'))
with live_query:
    next_ten_results = live_query.take(10)

Parameters

query (Query) – A query object

live_query_from_s(table_name: str, query: str) → tibco.liveview.models.live_query.LiveQuery

Prepare a live query from a string.

Usage:

import liveview

client = liveview.get_client('lv://localhost:11080')
query = client.live_query_from_s('ItemsSales', 'SELECT * FROM ItemsSales')
with live_query:
    next_ten_results = live_query.take(10)

Parameters

• table_name (str) – The name of the table to query.

• query (str) – The full LiveQL query to send.

Returns

Live query which may be listened to.

Return type

LiveQuery

snapshot_query(query: tibco.liveview.lv_query.Query) → tibco.liveview.models.snapshot_result.SnapshotResult

Issues a snapshot query to the server and returns the results.

Parameters

query (Query) – The query to run.

Returns

Snapshot result.

Return type

SnapshotResult

snapshot_query_from_s(table_name: str, query: str) → tibco.liveview.models.snapshot_result.SnapshotResult

Issues a snapshot query to the server and returns the results.

Parameters

• table_name (str) – The name of the table to query.

• query (str) – The full LiveQL query to send.

Returns

Snapshot result.

Return type

SnapshotResult

Query

class Query(table_name)

A Query allows you to build LiveQL expressions.

See LiveQL Reference for details.

Note that this class allows you to create invalid queries. It’s up to you to create valid queries.

Usage:

q = Query('ItemsSales').select('category', 'Item').where(category='electronics', Item='Surge Protector')
live_query = client.live_query(q)
# Or
# live_query = client.live_query_from_s(q.table_name, q.to_s())

between(timestamp1, timestamp2)

To be used with when(). Give two timestamps to define the bounds of this query’s time-windowed results.

Parameters

• timestamp1 – A timestamp which defines the beginning of a time-window

• timestamp2 – A timestamp which defines the end of a time-window

Returns

This Query object

Usage:

from liveview.util import now_timestamp

time1 = now_timestamp()
# ... Some logic during which time passes
time2 = now_timestamp()
query = Query('ItemsSales').select('Item').when('transactionTime').between(time1, time2)

for_ms(milliseconds: int)

Define a time-delay modifier.

See the Time-Based Data Modifiers section of the LiveQL Reference for details.

Parameters

milliseconds (int) – The number of milliseconds for which this Query’s predicate should hold

Returns

This Query object

group_by(*cols)

Group-by columns.

Parameters

*cols – One or more columns to group by.

Returns

This Query object

Usage:

query = Query('ItemsSales').select('category', 'COUNT(Item) AS itemCount').group_by('category')
# query.to_s() == "SELECT category, COUNT(Item) AS itemCount FROM ItemsSales GROUP BY category"

having(having_expr: str)

Provide a HAVING clause to this query.

Usage:

query = Query('ItemsSales').select('Item', 'category').order_by('category', Item='ASC').having('AVG(lastSoldPrice) > 5')
# query.to_s() == 'SELECT Item, category FROM ItemsSales ORDER BY category DESC, Item ASC HAVING AVG(lastSoldPrice) > 5'

Parameters

having_expr (str) – an aggregate HAVING expression

Returns

This Query object

limit(limit: int)

Limit the number of rows that this query will return.

Parameters

limit (int) – the max number of rows that this query should return

Returns

This Query object

order_by(*cols: str, **col_orders)

Define this query’s ordering

Usage:

# Default to descending order
# ORDER BY item DESC, transactionTime DESC
query.order_by('item', 'transactionTime')

# Or order by ascending order
# ORDER BY transactionTime ASC
query.order_by(transactionTime='ASC')

Parameters

• *cols (str) – column names which will default to ASC ordering

• **col_orders – column name keyword args which allow you to specify DESC with itemsSales='DESC'

Returns

This Query object

pivot(expr_list: List[str], values: List, for_col: Optional[str] = None, group_by: Union[str, List[str]] = [])

Have this query return results from a pivot table. See the Pivot Modifier section of the LiveQL Reference for details.

Parameters

• expr_list (List[str]) – A list of expressions which will fill the rows of the pivot table.

• values (List) – A list of values found in for_col, these values will be the column titles of the pivot table.

• for_col (str) – A column which contains values in the values list.

• group_by (Union[str, List[str]]) – a string or list of strings which are column names to group by.

Returns:

select(*projection: str)

Define a projection over this Query’s table.

If no arguments are given, the projection will default to ‘*’.

Parameters

*projection – one or more columns from the table

Returns

This Query object

Usage:

query = Query('ItemsSales').select('Item', 'category', 'transactionTime')

property table_name: str

The name of this query’s table

to_s() → str

Return LiveQL string built by this Query

Returns

This Query object as a string which LiveView can parse, if valid.

when(colname_timestamp: str)

Define a column for this query to provide time-windowed results around. To be used with between().

Parameters

colname_timestamp (str) – The name of a column around which a time-window will be defined

Returns

This Query object

Usage:

from liveview.util import now_timestamp

time1 = now_timestamp()
# ... Some logic during which time passes
time2 = now_timestamp()
query = Query('ItemsSales').select('Item').when('transactionTime').between(time1, time2)

where(str_predicate: str = '', **conj_predicates)

Define a predicate to filter rows from this Query.

Parameters

• str_predicate (str) –

  Optional; a string predicate like:

  "Item = 'This' OR Item = 'That' AND NOT category = 'automotive'"
  

• **conj_predicates –

  Keyword arg conjunctive predicates that will be ANDed together. For example:

  Query('CarSales').select('*').where(year=2021, country='France').to_s()
  

  will give you the string:

  "SELECT * FROM CarSales WHERE year = 2021 AND country = 'France'"
  

Returns

This Query object

Usage:

query1 = Query('ItemsSales').select('Item').where(category='Automotive')
# Or
query2 = Query('ItemsSales').select('Item').where("NOT category = 'Automotive'")

TableMetadata

class TableMetadata(tbl)

LiveView Table metadata. A TableMetadata object should not be instantiated

directly, but should be obtained through LiveViewClient#get_table_metadata().

- name

The name of the table

Type

str

- schema

The schema of the table

Type

str

- keys

The key(s) of the table

Type

list

Parameters

table_metadata (-) – The swagger table_info returned by a get_table request.

column_names() → List[str]

Get a list of all the column names of this table

get(field) → Optional[str]

Get the value of a field from this table’s metadata

to_dict() → dict

Returns the table as a dict, including all advanced fields.

Returns

this table’s metadata

Return type

dict

SnapshotResult

class SnapshotResult(snapshot)

The class is used to encapsulate the results of snapshot queries. Instantiated through LiveViewClient#snapshot_query().

Parameters

snapshot (-) – The raw JSON of a snapshot query, as a list.

- schema

Dict containing the Schema for the table queried.

- keys

List containing corresponding Key(s) for the tuples in ‘data’

- cols

List of the names of columns in the table

- snapshot_data

List of dicts, serving as the snapshot data.

- size

Number of tuples in the snapshot.

get_column(col_name) → list

Returns a list containing every entry for a given column.

Parameters

col_name (-) – The name of the column to get data from.

Returns

• A list of values.

to_dict() → dict

Returns the data of the snapshot as a dictionary.

Returns

• A Dict, with keys ‘schema’, ‘keys’, and ‘data’

LiveQuery

class LiveQuery(uri: str)

Container class for a LiveQuery. It’s recommended that you don’t create a LiveQuery directly, but instead follow the below usage.

Usage:

import liveview

# Assume Hello LiveView is running at lv://localhost:11080
liveview_server_uri = 'lv://localhost:11080'
client = liveview.get_client(liveview_server_uri)

with client.live_query_from_s('ItemsSales', 'SELECT * FROM ItemsSales') as live_query:
    results = live_query.take(20)

print(results)

close()

Close this query.

You don’t need to call this method if you use the with live_query syntax.

get_schema() → Dict

Get this query’s schema.

Returns (dict): A mapping from column names to data types.

iterator()

Get an iterator which yields results from this live query.

take(num_tuples: int) → List

Get the next num_tuples from this query. Blocks.

BasicQueryListener

class BasicQueryListener(live_query: tibco.liveview.models.live_query.LiveQuery, callback: Callable, error_callback: Optional[Callable] = None)

A BasicQueryListener listens for and processes any events from a live query.

Any non-error event received will be processed with the callback argument. Any error event received will be processed with the optional error_callback argument.

If no error_callback is provided, the listener will simply store the most recent exception as a member variable accessible through BasicQueryListener#get_error_result.

get_error_result()

The user-defined error_callback function may return a result. This method returns that result for access outside the event loop thread.

Returns: The last result returned by the error_callback function

start() → None

Start listening for events and calling the callback on incoming data.

Returns: None

stop() → None

Stop handling events and close the thread.

Returns: None

QueryListener

class QueryListener(live_query: tibco.liveview.models.live_query.LiveQuery, tuple_added: Optional[Callable] = None, tuple_removed: Optional[Callable] = None, tuple_updated: Optional[Callable] = None, delete_begin: Optional[Callable] = None, delete_end: Optional[Callable] = None, exception_raised: Optional[Callable] = None, snapshot_begin: Optional[Callable] = None, snapshot_end: Optional[Callable] = None)

A QueryListener listens to a live query and responds to events according to their type.

It provides no built-in callback functions and allows you to customize how events are handled. See liveview.listeners.LiveResult for an example of a QueryListener with callback functions implemented.

start() → None

Start listening for events and calling the callback on incoming data.

Returns: None

stop() → None

Stop handling events and close the thread.

Returns: None

LiveResult

class LiveResult(live_query: tibco.liveview.models.live_query.LiveQuery, exception_raised: Optional[Callable] = None, convert_with_schema: bool = True)

A LiveResult is a QueryListener that listens to a LiveQuery and maintains a result set of tuples.

It implements callback handlers for add, remove, and update events so that when LiveResult#get_columns() is called with desired columns as arguments, data from the current result set is retrieved.

Warning: The result set may grow infinitely. This class is only intended to be used with live queries

that produce finite result sets.

Usage:

import liveview as lv
import time
from liveview.listeners import LiveResult

# Assume a LiveView server is running 'Hello LiveView' locally on port 11080
client = liveview.get_client('lv://localhost:11080')
live_query = client.live_query_from_s('ItemsSales', 'SELECT * FROM ItemsSales')
with LiveResult(live_query) as live_result:
    time.sleep(3)
categories, items = live_result.get_columns('category', 'Item')

get_columns(*columns) → List[List[any]]

Get columns of live data as a list of lists.

Usage:

temperatures, room_names = live_result.get_columns('Temperature', 'Room')
# temperatures == [70.2, 105.3, 75.3]
# room_names == ['Kitchen', 'Sauna', 'Living Room']

# Get a single column, unpack the list of lists.
temperatures, *_ = live_result.get_columns('Temperature')

Parameters

*columns – one or more tuple field names

Returns

A list of the columns in the same order as requested in the arguments

get_tuples() → Dict[int, Dict[str, any]]

Get a mapping from tuple primary keys to tuple data

BufferedLVPublisher

class BufferedLVPublisher(api_client: tibco.liveview.liveview_generated_client.swagger_client.api.v1_1_api.V11Api, host: str, table_name: str, publish_interval: float = 1)

Asynchronously publishes tuples to a LiveView table using a FIFO buffer. Initialize with:

import liveview
client = liveview.get_client('lv://localhost:11080')
publisher = client.get_buffered_publisher(table_name='ItemsSales', publish_interval=1)

buffer_size() → int

Return the number of tuples currently in the publish queue

flush_buffer() → List

Empty out the buffer and send it to the server.

Useful for publishing tuples at the end of a program, when publish_interval seconds may not pass before Python exits.

Returns

List of tuples published

publish(lv_tuple)

Insert a tuple or list of tuples into the buffer to be published when this object’s publish interval

start() → None

Publish any incoming tuples at an interval. Called when object is initialized.

stop() → None

Stop the publisher from automatically publishing at an interval.

Util

convert_tuple_with_schema(liveview_tuple: dict, schema: dict) → dict

Try to convert a LiveView tuple’s fields into their appropriate data types according to a schema. The schema is a dictionary that maps from string field names to string LiveView Type names from the below table.

Fields are mapped to Python types according to the below table. If a conversion is unable to be performed, the field will be present in this function’s return value without conversion.

LiveView Type	Python Type
blob	bytes
bool	bool
double	float
int	int
long	int
string	str
timestamp	datetime

timestamp_to_datetime(ts) → datetime.datetime

Convert an absolute timestamp in milliseconds to a Python datetime.

now_timestamp() → int

Return an absolute timestamp in milliseconds for the present moment.