Working with metadata (table definitions etc.)

You can access metadata and update a table definition such as the query of a table.

Examples:

# Get a list of all databases in your account.
# The result will include all tables per DB and all fields (columns) per table.
databases = pq.list_databases()
st.json(databases)

# Generate a UNION SQL query for a set of tables.
# Sources is optional, it adds an extra column in the UNION to indicate
# the source of the records per table.
table_ids = [1, 2, 7]
source_names = { 1: "Paris", 2: "London", 7: "Gent" }
union = pq.generate_sql_union(table_ids = table_ids, sources = source_names)
st.text(union)

# Update a table, e.g. set a new query for a table.
# Note: in this example table id 123 must exist and should have table_type="query".
pq.update_table(id = 123, query = union)

# Update a database, e.g. set the description (data catalog meta data)
pq.update_database(id = 123, description = "Sales orders from accounting")

# Update a column (field), e.g. set the description (data catalog meta data)
# Do not confuse with update_cell which is used to update one cell of one record
pq.update_field(id = 123, description = "Curreny of the order")

Generate a UNION query dynamically

These functions allow you to dynamically generate e.g. a UNION query that includes a list of tables from your account. The app can be scheduled to run daily, in order to keep the UNION query up to date as new tables are added.

Here’s an example that loops over all tables in all databases, filters on table name and builds up a UNION query that includes all the matching tables:

databases = pq.list_databases()

table_ids = []
source_names = {}
for db in databases:
  for table in db["tables"]:
    if "<some word>" in table["name"]:
      table_ids.append(table["id"])
      source_names[table["id"]] = table["name"].replace("<some word>", "")

union = pq.generate_sql_union(table_ids = table_ids, sources = source_names)
pq.update_table(id = 123, query = union)

Import metadata from an OpenAPI specification

Here’s an example on how to extract column descriptions from an OpenAPI JSON specification and insert into the column (field) descriptions in Peliqan:

import json

def main():  
  field_descriptions = get_field_descriptions()
  set_field_descriptions(field_descriptions)

def get_field_descriptions():
  openapi_spec = """ {
	    "openapi": "3.0.0",
	    "components": {
	        "schemas": {
	            "Order": {
	                "allOf": [ {
	                        "type": "object",
	                        "properties": {
	                            "id": {
	                                "type": "integer",
	                                "description": "Id of the order"
	                            },
	                            "name": {
	                                "type": "string",
	                                "description": "Name of the order"
	} } } ] } } } }
  """
  openapi_schema = json.loads(openapi_spec)
  field_descriptions = {}
  schemas = openapi_schema["components"]["schemas"]
  for objectName, schema in schemas.items():
    field_descriptions[objectName.lower()] = {}
    for schemaItem in schema["allOf"]:
      if "properties" in schemaItem:
        for propertyName, property in schemaItem["properties"].items():
          if "description" in property:
            field_descriptions[objectName.lower()][propertyName.lower()] = property["description"]
  return field_descriptions

def set_field_descriptions(field_descriptions):
  databases = pq.list_databases()
  for db in databases:
    if db["name"] == 'accounting':
      for table in db["tables"]:
        tableName = table["name"].lower().replace("public.", "")
        if tableName in field_descriptions:
          for field in table["all_fields"]:
            fieldName = field["name"].lower()
            if fieldName in field_descriptions[tableName]:
              pq.update_field(id = field["id"], description = field_descriptions[tableName][fieldName])

main()