Reference - Peliqan documentation

Data apps are interactive apps, built in single-file Python scripts, to run on Peliqan. This page provides a Reference for building Data apps.

Data apps on the Peliqan platform

Peliqan is an all-in-one data platform with a wide range of components, that can be controlled and used from within Data apps using Peliqan’s pq module:

Connectors (ETL pipelines that sync data from sources into a datawarehouse)
Built-in datawarehouse, and ability to connect an external datawarehouse as ETL target
Data explorer & SQL queries for data transformations (queries can be configured as views and/or can be materialized)
Trino federated query engine
Data replication
Python IDE and runtime to develop, test, host, schedule & publish Data apps

Peliqan pq module

The pq module exposes Peliqan’s low-code data-centric and data pipeline functions. The pq module is automatically available in Peliqan when you write Python code in Data apps and can be used in your local IDE by doing pip install peliqan and using import peliqan as pq in your code.

Constants

# Run modes

Data apps can run locally on a computer (external to Peliqan) or inside the Peliqan platform. When hosted on Peliqan, some code is prepended, e.g. to provide context and to automatically import Streamlit as st and Peliqan as pq. Constant RUN_CONTEXT is always set when running a Data app in Peliqan. Data apps can run interactively or in the background (e.g. scheduled). Streamlit is used to build UI’s for interactive apps. Streamlit functions such as st.text(), st.write(), st.json(), st.table() etc. are converted to print statements in background runs. Print statements are written to the log file of each run, visible in Peliqan.

Data warehouse, databases and tables

Connect to data warehouse or DB

Read

Insert

dbconn.insert('db_name', 'schema_name', 'table_name', record_dict)

Update

dbconn.update('db_name', 'schema_name', 'table_name', 'row_pk_value', record_dict)

Upsert

Inserts or updates based on the row_pk_value.

dbconn.upsert('db_name', 'schema_name', 'table_name', 'row_pk_value', record_dict)

Write

Write implements full ELT pipeline logic: cast to schema (if given), upsert (insert or update based on pk value), and creates table and columns if needed.

Execute query

dbconn.execute('db_name', query='TRUNCATE TABLE schema_name.table_name')

Federated query engine Trino

Connect Trino

trinoconn = pq.trinoconnect()

Read

rows = trinoconn.fetch('db_or_catalog_name', 'schema_name', 'table_name')

# Read data using custom query
rows = trinoconn.fetch('db_name', query='SELECT * FROM catalog.schema.table_name')

# Get result as a data frame
df = trinoconn.fetch('db_name', 'schema_name', 'table_name', df=True)

Writeback (SaaS APIs)

Writeback functions are wrappers for the APIs of connected SaaS applications. Some SaaS connections in Peliqan are sources only (ELT pipelines), others are writeback only and some SaaS connections are both.

Writeback examples:

More info: Writeback to SaaS APIs

SFTP

Connect SFTP

sftpconn = pq.sftpconnect('connection_name')

List files

result = sftpconn.dir('.')

Read file

result = sftpconn.read_file('./file.json')

State

State is stored and maintained across app runs. For session-state (within the scope of a single run), use Streamlit’s session state: st.session_state().

Get state

state = pq.get_state()

if not state:
	state={}

settings = state.get("settings", {})

Set state

new_state = {
	"settings": { ... }
}

pq.set_state(new_state)

Sharing state between two or more Data Apps running on Peliqan (e.g. in an orchestration pipeline):

Sharing state between a Data App running locally on your computer, and a Data App on Peliqan:

Note: when running a Data App outside of Peliqan, by default pq.get_state() and pq.set_state(state)will read and write the state locally in a temporary file. The above code however, shares the state with a Data App running on Peliqan. Make sure that PELIQAN_INTERFACE_ID is set as a local environment variable, with the id of the Data App on Peliqan.

Metadata

List databases

databases = pq.list_databases()
st.json(databases)

List schemas

schemas = pq.list_schemas(db_name)
st.json(schemas)

List tables

databases = pq.list_databases()
for db in databases:
	for table in db['tables']:
		st.text(table['name'])

List scripts (data apps)

scripts = pq.list_scripts()
st.json(scripts)

Get database

db_meta_data = pq.get_database(database_id=37734)
st.json(db_meta_data)

Get schema

schema_meta_data = pq.get_schema(schema_id=37734)
st.json(db_meta_data)

Get table

table_meta_data = pq.get_table(table_id=37734)
st.json(table_meta_data)

Get script

script_meta_data = pq.get_script(script_id=745)
st.json(script_meta_data)

Update database

db_meta_data = pq.update_database(database_id=1603, description="This is a database description.")
st.json(db_meta_data)

Update schema

schema_meta_data = pq.update_schema(schema_id=1603, name="my_schema")
st.json(schema_meta_data)

Add table

dbconn = pq.dbconnect(pq.DW_NAME)

fields = {
    "id": "integer",
    "name": "text",
    "dob": "date_time"
}

result = dbconn.create_table(db_name=pq.DW_NAME, schema_name='my_data', table_name='my_new_table', fields=fields, pk='id')

Update table

table_meta_data = pq.update_table(table_id=37734, description="This is a table description.")
st.json(table_meta_data)

Update field

field_meta_data = pq.update_field(field_id=829137, description="This is a field description.")
st.json(field_meta_data)

Update script

Add script

pq.add_script(group_id=123, name='my_new_script', raw_script='...')
pq.add_script(group_name='Production data', name='my_new_script', raw_script='...')
pq.add_script(group_name='Production data', name='my_new_script', raw_script='...', run_mode=STREAMLIT)

run_mode supported values are STREAMLIT, API, SHELL.

Add or update a query

pq.add_query(name='My query', query='SELECT * from odoo.accounts', as_view=False)
pq.upsert_query(name='My query', query='SELECT * from odoo.accounts', schema_id=123)

Replicate a table or materialize a query

Manage Data App templates

Templates are managed in a parent account (partner account) and used in sub accounts to create Data Apps (instances, copies of the template).

Pipelines

Run pipeline

Run pipeline for all active tables in the connection:

pq.run_pipeline(connection_name = 'Hubspot', is_async = False)

Run pipeline for one table:

pq.run_pipeline(connection_name = 'Hubspot', tables = "deals")

Run pipeline for a list of tables:

pq.run_pipeline(connection_name = 'Hubspot', tables = ["deals", "contacts"])

Run pipeline with a specific parameter (overrides connection form):

pq.run_pipeline(connection_name = 'Hubspot', tables = ["deals"], param1 = 100)

Example where param1 is the “pagesize” in the connection form.

Get pipeline runs

pq.get_pipeline_runs(source_connection_id)

Get pipeline logs

pq.get_pipeline_logs(pipeline_run_id)

Secret store

Store credential and other secrets securely (in Peliqan > Connections > Add connection > Secret Store).

pq.get_secret(secret_name_or_id)

Refresh

A refresh updates the metadata in Peliqan for a connected database, meaning getting an up to date list of e.g. tables in a database.

Refresh database

pq.refresh_database(database_name = 'dw_12345', is_async = False)

Refresh schema

pq.refresh_schema(database_name = 'dw_12345', schema_name = 'CRM', is_async = False)

Refresh table

pq.refresh_table(database_name = 'dw_12345', schema_name = 'CRM', table_name = 'customers', is_async = False)

Sub accounts

Manage sub accounts from a parent (partner) account.

List sub accounts

result = pq.list_subaccounts()

Get sub account

result = pq.get_subaccount(account_id=1234)
result = pq.get_subaccount(external_id='1234124-134412')

Add sub account

Get sub account pq instance

pq_sub_account = pq.get_subaccount_instance(account_id=1234)
connections_in_sub_account = pq_sub_account.list_connections() # Example: get the connections in the sub account

Get sub account API token

result = pq.get_subaccount_token(account_id=1234)
result = pq.get_subaccount_token(external_id='1234124-134412')

Get sub account info

result = pq.get_subaccount_info(account_id=1234)

General

Generic API call to Peliqan backend

Example update field (column) in table:

method = 'post'
url = f"{pq.BACKEND_URL}/api/database/fields/{field_id}/"
kwargs={
    'json': {
        'type': 'link_row',
        'fk_table_id': linked_table_id,
        'name': field_name
    }
}
pq.__service_client__.call_backend(method=method, url=url, expected_status_code=200, **kwargs)

Useful code snippets

Code to run data apps locally (outside of Peliqan) as well as on Peliqan, without any required code changes:

if 'RUN_CONTEXT' in globals(): # Running on Peliqan
    RUN_ENV = 'peliqan'
else: # Running outside of Peliqan
    RUN_ENV = 'local'
    from peliqan import Peliqan
    import streamlit as st
    import os
    api_key = os.getenv("PELIQAN_API_KEY")
    if not api_key:
        st.error("PELIQAN_API_KEY environment variable is not set.")
        st.stop()
    interface_id = os.getenv("PELIQAN_INTERFACE_ID", 0) # script id in Peliqan, used to share state (see below)
    pq = Peliqan(api_key)
    try: # Check if running locally in Streamlit
        from streamlit.runtime.scriptrunner import get_script_run_ctx
        if get_script_run_ctx() is not None:
            RUN_CONTEXT = "interactive"
    except Exception:
        RUN_CONTEXT = "background"

# Other code goes here

List all databases, schemas, tables in current account:

def get_tables():
    """Get all tables from all databases in current account"""
    all_tables = []
    for db in pq.list_databases():
        for table in db["tables"]:
            schema_name = next((s["name"] for s in db["schemas"] if s["id"] == table["schema_id"]), None)
            all_tables.append({
                "db_id": db["id"],
                "schema_id": table["schema_id"],
                "table_id": table["id"],
                "db": db["name"],
                "schema": schema_name,
                "table": table["name"]
            })
    return all_tables

Get table fields & PK:

def get_table_columns_and_pk(table_id):
    """Get all columns and primary key from a table"""
    table_meta = pq.get_table(table_id)
    columns = table_meta.get("all_fields", [])
    text_columns = [col["name"] for col in columns
                   if col.get("type") in ["text", "varchar", "string"]
                   and not col["name"].startswith("_sdc")]
    pk = next((col["name"] for col in columns if col.get("is_pk")), "id")
    return text_columns, pk

Apply Peliqan’s look & feel in Streamlit interactive Data apps:

def apply_styling():
    st.markdown("""
	    <link rel="preconnect" href="https://fonts.googleapis.com">
	    <link rel="preconnect" href="https://fonts.gstatic.com" crossorigin>
	    <link href="https://fonts.googleapis.com/css2?family=Asap:ital,wght@0,100..900;1,100..900&family=Roboto:ital,wght@0,100..900;1,100..900&display=swap" rel="stylesheet">
	    <link href="https://fonts.googleapis.com/css2?family=Material+Symbols+Outlined" rel="stylesheet"/>
	    
	    <style>
	        html, body, h4, [class*="st-"]:not([data-testid="stIconMaterial"]), .main, .stApp {
	            font-family: 'Roboto', sans-serif !important;
	            font-size: 14px !important;
	        }
	        
	        #stDecoration {
	            display: none;
	        }
	        
	        h1, .main-header {
	            font-family: 'Roboto', sans-serif !important;
	            font-size: 28px !important;
	            color: #1f77b4;
	            margin-top: 1rem;
	            margin-bottom: 1rem;
	        }
	                
	        h2, .sub-header {
	            font-family: 'Roboto', sans-serif !important;
	            font-size: 24px !important;
	            color: #1f77b4;
	            margin-top: 1rem;
	            margin-bottom: 1rem;
	        }
	                
	        h3 {
	            font-family: 'Roboto', sans-serif !important;
	            font-size: 20px !important;
	            color: #1f77b4;
	            margin-top: 1rem;
	            margin-bottom: 1rem;
	        }
	                
	        .stMainBlockContainer {
	            padding-top: 1.5rem;
	            padding-bottom: 1rem;
	        }
	        
	        
	    </style>
	    """,
        unsafe_allow_html=True,
    )

	st.set_page_config(
	    page_title="Peliqan Data App",
	    page_icon="",
	    layout="wide",
	    initial_sidebar_state="auto"
	)

Logging:

LOG_LEVEL = 100
def print_log(msg: str, level: int = 1) -> None:
    if level < LOG_LEVEL:
        caller_frame = inspect.currentframe().f_back
        line_no = caller_frame.f_lineno
        print(f"{line_no} \u2192 {msg}")

API handler Data app example:

def handler(request):
	# Get request details:
	method = request['method']
  url = request['url']
  path = request['path']
  query_string = request['query_string']
  headers = request['headers']
  data = request['data']
  form = request['form']
	
	# Example reading a parameter from the querystring
	from urllib.parse import parse_qs
  query_string_dict = parse_qs(query_string)
  if "param1" in query_string_dict:
	  param1 = query_string_dict['param1'][0]

  # Example reading data from an incoming POST body with JSON
  import json
  postbody_obj = json.loads(data)
  
	# Example response: read rows from a table (pq = Peliqan module)
	dbconn = pq.dbconnect('dw_2')
	table_rows = dbconn.fetch('dw_2', 'salesforce', 'accounts')
	
	# Send the API response (e.g. return the table contents)
	return table_rows

pq.DW_NAME       # Name of default data warehouse (database) in the current account, also name of the default data warehouse connection
pq.DW_ID         # Id of default data warehouse in the current account
pq.INTERFACE_ID  # Current script id
RUN_CONTEXT      # Possible values: interactive, scheduled, api_handler or not set (outside of Peliqan)

# Example detect if script running inside or outside of Peliqan:

if 'RUN_CONTEXT' not in globals():
	print("Running outside of Peliqan)

# Get the current logged-in user
user = pq.__service_client__.call_backend(method="get", url=f"{pq.BACKEND_URL}/api/user/me")
st.write(user["username"])

# Connect to e.g. a SQL Server, MySQL, Postgres DB added in Peliqan under "Connections"
dbconn = pq.dbconnect('connection_name')

# Connect to built-in default data warehouse
dbconn = pq.dbconnect(pq.DW_NAME)

# Use the connection to read data
rows = dbconn.fetch('db_name', 'schema_name', 'table_name')

# Read data from default built-in data warehouse
rows = dbconn.fetch(pq.DW_NAME, 'schema_name', 'table_name')

rows = dbconn.fetch('db_name', 'schema_name', 'table_name')
rows = dbconn.fetch('db_name', 'schema_name', 'table_name')

# Read data using custom query
rows = dbconn.fetch('db_name', query='SELECT * FROM schema_name.table_name')

# Get result as a data frame
df = dbconn.fetch('db_name', 'schema_name', 'table_name', df=True)

# Options:
# df = True: set to True to return a Dataframe instead of a list of objects
# fillna_with = '': replace Na (not a number) with a given value, e.g. in empty string
# fillnat_with= '': replace NaT (not a date)
# enable_python_types = True: use Python types in response
# enable_datetime_as_string = True: return datetime columns as string
# tz='UTC': timezone for datetimes

dbconn.write('schema_name', 'table_name', [{'id': 1, 'name': 'John'}], pk='id')

# Write with a schema provided
object_schema = {
	'properties': {
		'id': {'type': 'integer'},
		'name': {'type': 'string'}
	}
}
dbconn.write('schema_name', 'table_name', [{'id': 1, 'name': 'John'}], object_schema, pk='id')

apiconn = pq.connect('connection_name')

apiconn.add('object_type', object)   # or apiconn.add('object_type', **object)

apiconn.update('object_type', object)

apiconn.upsert('contact', contact, searchfield='email', searchterm='john@doe.com')

apiconn.delete('object_type', id=1)

apiconn.get('object_type', id=1)

apiconn.findone('object_type', searchfield='email', searchterm='john@doe.com')

apiconn.list('object_type')

apiconn.copy('object_type', id=1)

apiconn.rename('object_type', id=1, name='new name')

apiconn.apicall('path', method='get', body={})

# Send message to a Slack channel
slack_api = pq.connect('Slack')
message = {
    'text': "Data quality alert",
    'channel': "Alerts",
    'username': "The Peliqan bot",
}
result = slack_api.add('message', message)

# Odoo add partner
odoo_api = pq.connect('Odoo')
partner = {
    'name': "ACME",
    'email': "info@acme.com",
}
result = odoo_api.add('partner', partner)

def get_state_from_app(app_id): 
    url = f"{pq.BACKEND_URL}/api/interfaces/{app_id}/state/"
    result = pq.__service_client__.call_backend(method="get", url=url, expected_status_code=200)
    state = result["state"]
    return state

def set_state_for_app(app_id, state):
    url = f"{pq.BACKEND_URL}/api/interfaces/{app_id}/state/"
    payload = {"state": state}
    result = pq.__service_client__.call_backend(method="post", url=url, json=payload, expected_status_code=200)

state = get_state_from_app(1234)
state["new_key"] = 555
set_state_for_app(5678, state)

script_meta_data = pq.update_script(script_id=745, name="Create and hydrate subaccount")
st.json(script_meta_data)

# Set a schedule
settings = {
    "schedule_settings": {
        "schedule": True,
        "weekdays": [0, 1, 2, 3, 4, 5, 6],
        "task_name": f"run_python_app_{script_id}",
        "start_time": start_time,
        "run_interval": 60*60*24, # once every day
    }
}
script_meta_data = pq.update_script(script_id=script_id, settings=settings)["settings"]["schedule_settings"]

# Example one-time replicate to built-in DWH
dbconn = pq.dbconnect(pq.DW_NAME) # e.g. dw_123 (connection name & DWH name)
result = dbconn.replicate(db_name=pq.DW_NAME, table_source_list=[ (source_table_id, target_table_name), (source_table_id, target_table_name) ], target_schema='my_target_schema')
st.write(result)


# Example one-time replicate to external target
dbconn = pq.dbconnect('MS SQL Server')
result = dbconn.replicate(db_name='my_target_db', table_source_list=[ (source_table_id, target_table_name), (source_table_id, target_table_name) ], target_schema='my_target_schema')
st.write(result)


# Example scheduled replicate to built-in DWH
replicate_settings = {
  "schedule": {
    "interval": 3600,                                    # 3600 (hourly), 21600 (every 6 hours), 86400 (daily)
    "weekdays": [0,1,2,3,4,5,6],                         # 0 = Monday
    "start_time": "03:30:00"                             # Only used when interval = 86400 (24 hour interval)
  }
  "source_incremental_field": "timestamp_last_update",   # optional field, value should be an ISO 8601 timestamp
  "target_table_name":        "replicated_report",       # required field
  "target_schema_name":       "my_replicated_data",      # required field
  "target_database_id":       pq.DW_ID,                  # required field
  "target_connection_name":   pq.DW_NAME,                # required field
}
result = pq.replicate(table_id=source_table_id, enabled=True, settings=replicate_settings)
st.write(result)


# Example scheduled replicate to external target
replicate_settings = {
  "schedule": {
    "interval": 3600,                                   # 3600 (hourly), 21600 (every 6 hours), 86400 (daily)
    "weekdays": [0,1,2,3,4,5,6],                        # 0 = Monday
    "start_time": "03:30:00"                            # Only used when interval = 86400 (24 hour interval)
  }
  "source_incremental_field": "timestamp_last_update",  # optional field, value should be an ISO 8601 timestamp
  "target_table_name":        "replicated_report",      # required field
  "target_schema_name":       "my_replicated_data",     # required field
  "target_database_id":       123,                      # required field
  "target_connection_name":   "SQL Server",             # required field
}
result = pq.replicate(table_id=source_table_id, enabled=True, settings=replicate_settings)
st.write(result)


# Replicate/materialize to a subaccount
result = pq.subaccount_materialize(connection_name = 'Hubspot', subaccount_id = 123, table_source_list = ['contacts'], target_schema = 'materialized_queries')
st.write(result)

# list templates
url = f"{pq.BACKEND_URL}/api/partner/interface-templates/"
result = pq.__service_client__.call_backend(method='get', url=url, expected_status_code=200)
st.write(result)

# get template
template_id = 123
url = f"{pq.BACKEND_URL}/api/partner/interface-templates/{template_id}/"
result = pq.__service_client__.call_backend(method='get', url=url, expected_status_code=200)
st.write(result)

# update template
template_id = 123
raw_script = "st.write('first line')\nst.write('second line')"
url = f"{pq.BACKEND_URL}/api/partner/interface-templates/{template_id}/"
json = {"raw_script": raw_script}
result = pq.__service_client__.call_backend(method='patch', url=url, expected_status_code=200, json=json)
st.write(result)

account_owner_details={
  'first_name': 'first_name',
  'last_name': 'last_name',
  'email': 'email@company.com',
  'password': 'password'
}
company_details={
  'company_name': 'company_name',
  'country': 'country',
  'company_size': ''
}
pq.add_subaccount(account_owner_details, company_details)

​Data apps on the Peliqan platform

​Peliqan pq module

​Constants

​Data warehouse, databases and tables

​Connect to data warehouse or DB

​Read

​Insert

​Update

​Upsert

​Write

​Execute query

​Federated query engine Trino

​Connect Trino

​Read

​Writeback (SaaS APIs)

​SFTP

​Connect SFTP

​List files

​Read file

​State

​Get state

​Set state

​Sharing state between apps

​Metadata

​List databases

​List schemas

​List tables

​List scripts (data apps)

​Get database

​Get schema

​Get table

​Get script

​Update database

​Update schema

​Add table

​Update table

​Update field

​Update script

​Add script

​Add or update a query

​Replicate a table or materialize a query

​Manage Data App templates

​Pipelines

​Run pipeline

​Get pipeline runs

​Get pipeline logs

​Secret store

​Refresh

​Refresh database

​Refresh schema

​Refresh table

​Sub accounts

​List sub accounts

​Get sub account

​Add sub account

​Get sub account pq instance

​Get sub account API token

​Get sub account info

​General

​Generic API call to Peliqan backend

​Useful code snippets

Data apps on the Peliqan platform

Peliqan pq module

Constants

Data warehouse, databases and tables

Connect to data warehouse or DB

Read

Insert

Update

Upsert

Write

Execute query

Federated query engine Trino

Connect Trino

Read

Writeback (SaaS APIs)

SFTP

Connect SFTP

List files

Read file

State

Get state

Set state

Sharing state between apps

Metadata

List databases

List schemas

List tables

List scripts (data apps)

Get database

Get schema

Get table

Get script

Update database

Update schema

Add table

Update table

Update field

Update script

Add script

Add or update a query

Replicate a table or materialize a query

Manage Data App templates

Pipelines

Run pipeline

Get pipeline runs

Get pipeline logs

Secret store

Refresh

Refresh database

Refresh schema

Refresh table

Sub accounts

List sub accounts

Get sub account

Add sub account

Get sub account pq instance

Get sub account API token

Get sub account info

General

Generic API call to Peliqan backend

Useful code snippets