import logging
from parsons.etl.table import Table
from parsons.utilities import check_env
from parsons.utilities.api_connector import APIConnector
logger = logging.getLogger(__name__)
[docs]class Quickbase(object):
"""
Instantiate the Quickbase class
`Args:`
hostname: str
The URL for the homepage/login page of the organization's Quickbase
instance (e.g. demo.quickbase.com).
user_token: str
The Quickbase account user token (API key). Not required if
``QUICKBASE_USER_TOKEN`` env variable is set.
`Returns:`
Quickbase Class
"""
def __init__(self, hostname=None, user_token=None):
self.hostname = check_env.check("QUICKBASE_HOSTNAME", hostname)
self.user_token = check_env.check("QUICKBASE_USER_TOKEN", user_token)
self.api_hostname = "https://api.quickbase.com/v1"
self.client = APIConnector(
self.api_hostname,
headers={
"QB-Realm-Hostname": self.hostname,
"AUTHORIZATION": f"QB-USER-TOKEN {self.user_token}",
},
)
[docs] def get_app_tables(self, app_id=None):
"""
Query records in a Quickbase table. This follows the patterns laid out
in Quickbase query documentaiton, located here:
https://help.quickbase.com/api-guide/componentsquery.html
`Args:`
app_id: str
Identifies which Quickbase app from which to fetch tables.
`Returns:`
Table Class
"""
return Table(
self.client.request(
f"{self.api_hostname}/tables?appId={app_id}", "GET"
).json()
)
[docs] def query_records(self, table_from=None):
"""
Query records in a Quickbase table. This follows the patterns laid out
in Quickbase query documentaiton, located here:
https://help.quickbase.com/api-guide/componentsquery.html
`Args:`
from: str
The ID of a Quickbase resource (i.e. a table) to query.
`Returns:`
Table Class
"""
req_resp = self.client.request(
f"{self.api_hostname}/records/query", "POST", json={"from": table_from}
).json()
resp_tbl = Table(req_resp["data"])
cleaned_tbl = Table()
for row in resp_tbl:
row_dict = {}
for column in resp_tbl.columns:
row_dict[column] = row[column]["value"]
cleaned_tbl.concat(Table([row_dict]))
cleaned_tbl.materialize()
column_resp = req_resp["fields"]
column_map = {}
for entry in column_resp:
column_map[str(entry["id"])] = entry["label"].lower().strip()
for column in cleaned_tbl.columns:
cleaned_tbl.rename_column(column, column_map[column])
return cleaned_tbl