> ## Documentation Index > Fetch the complete documentation index at: https://docs.peliqan.io/llms.txt > Use this file to discover all available pages before exploring further. # Reference > Complete reference guide for building Peliqan Data apps using the pq module, covering data warehouse operations, Trino queries, writeback, SFTP, state management, metadata APIs, pipelines, secrets, refresh, sub accounts, and useful code snippets. 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 ```python theme={null} # 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 ```python theme={null} ``` ### Read ```python theme={null} ``` ### Insert ```python theme={null} dbconn.insert('db_name', 'schema_name', 'table_name', record_dict) ``` ### Update ```python theme={null} dbconn.update('db_name', 'schema_name', 'table_name', 'row_pk_value', record_dict) ``` ### Upsert Inserts or updates based on the `row_pk_value`. ```python theme={null} 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. ```python theme={null} ``` ### Execute query ```python theme={null} dbconn.execute('db_name', query='TRUNCATE TABLE schema_name.table_name') ``` ## Federated query engine Trino ### Connect Trino ```python theme={null} trinoconn = pq.trinoconnect() ``` ### Read ```python theme={null} 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. ```python theme={null} ``` Writeback examples: ```python theme={null} ``` More info: [Writeback to SaaS APIs](/low-code-python-data-apps/writeback-reverse-etl-data-sync) ## SFTP ### Connect SFTP ```python theme={null} sftpconn = pq.sftpconnect('connection_name') ``` ### List files ```python theme={null} result = sftpconn.dir('.') ``` ### Read file ```python theme={null} 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 ```python theme={null} state = pq.get_state() if not state: state={} settings = state.get("settings", {}) ``` ### Set state ```python theme={null} new_state = { "settings": { ... } } pq.set_state(new_state) ``` ### Sharing state between apps Sharing state between two or more Data Apps running on Peliqan (e.g. in an orchestration pipeline): ```python theme={null} ``` Sharing state between a Data App running locally on your computer, and a Data App on Peliqan: ```python theme={null} ``` 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 ```python theme={null} databases = pq.list_databases() st.json(databases) ``` ### List schemas ```python theme={null} schemas = pq.list_schemas(db_name) st.json(schemas) ``` ### List tables ```python theme={null} databases = pq.list_databases() for db in databases: for table in db['tables']: st.text(table['name']) ``` ### List scripts (data apps) ```python theme={null} scripts = pq.list_scripts() st.json(scripts) ``` ### Get database ```python theme={null} db_meta_data = pq.get_database(database_id=37734) st.json(db_meta_data) ``` ### Get schema ```python theme={null} schema_meta_data = pq.get_schema(schema_id=37734) st.json(db_meta_data) ``` ### Get table ```python theme={null} table_meta_data = pq.get_table(table_id=37734) st.json(table_meta_data) ``` ### Get script ```python theme={null} script_meta_data = pq.get_script(script_id=745) st.json(script_meta_data) ``` ### Update database ```python theme={null} db_meta_data = pq.update_database(database_id=1603, description="This is a database description.") st.json(db_meta_data) ``` ### Update schema ```python theme={null} schema_meta_data = pq.update_schema(schema_id=1603, name="my_schema") st.json(schema_meta_data) ``` ### Add table ```python theme={null} 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 ```python theme={null} table_meta_data = pq.update_table(table_id=37734, description="This is a table description.") st.json(table_meta_data) ``` ### Update field ```python theme={null} field_meta_data = pq.update_field(field_id=829137, description="This is a field description.") st.json(field_meta_data) ``` ### Update script ```python theme={null} ``` ### Add script ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} ``` ### 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). ```python theme={null} ``` ## Pipelines ### Run pipeline Run pipeline for all active tables in the connection: ```python theme={null} pq.run_pipeline(connection_name = 'Hubspot', is_async = False) ``` Run pipeline for one table: ```python theme={null} pq.run_pipeline(connection_name = 'Hubspot', tables = "deals") ``` Run pipeline for a list of tables: ```python theme={null} pq.run_pipeline(connection_name = 'Hubspot', tables = ["deals", "contacts"]) ``` Run pipeline with a specific parameter (overrides connection form): ```python theme={null} pq.run_pipeline(connection_name = 'Hubspot', tables = ["deals"], param1 = 100) ``` Example where param1 is the "pagesize" in the connection form. ### Get pipeline runs ```python theme={null} pq.get_pipeline_runs(source_connection_id) ``` ### Get pipeline logs ```python theme={null} pq.get_pipeline_logs(pipeline_run_id) ``` ## Secret store Store credential and other secrets securely (in Peliqan > Connections > Add connection > Secret Store). ```python theme={null} 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 ```python theme={null} pq.refresh_database(database_name = 'dw_12345', is_async = False) ``` ### Refresh schema ```python theme={null} pq.refresh_schema(database_name = 'dw_12345', schema_name = 'CRM', is_async = False) ``` ### Refresh table ```python theme={null} 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 ```python theme={null} result = pq.list_subaccounts() ``` ### Get sub account ```python theme={null} result = pq.get_subaccount(account_id=1234) result = pq.get_subaccount(external_id='1234124-134412') ``` ### Add sub account ```python theme={null} ``` ### Get sub account pq instance ```python theme={null} 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 ```python theme={null} result = pq.get_subaccount_token(account_id=1234) result = pq.get_subaccount_token(external_id='1234124-134412') ``` ### Get sub account info ```python theme={null} result = pq.get_subaccount_info(account_id=1234) ``` ## General ### Generic API call to Peliqan backend Example update field (column) in table: ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} 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: ```python theme={null} def apply_styling(): st.markdown(""" """, unsafe_allow_html=True, ) st.set_page_config( page_title="Peliqan Data App", page_icon="", layout="wide", initial_sidebar_state="auto" ) ``` Logging: ```python theme={null} 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: ```python theme={null} 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 ``` ```python theme={null} 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"]) ``` ```python theme={null} # 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') ``` ```python theme={null} 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 ``` ```python theme={null} 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') ``` ```python theme={null} 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={}) ``` ```python theme={null} # 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) ``` ```python theme={null} 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) ``` ```python theme={null} 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"] ``` ```python theme={null} # 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) ``` ```none theme={null} # 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) ``` ```python theme={null} 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) ``` This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.