kevinveenbirkenbach/baserow-ifto
1
0
Fork 0
mirror of https://github.com/kevinveenbirkenbach/baserow-ifto.git synced 2024-11-22 19:01:05 +01:00
Code Issues Packages Projects Releases Wiki Activity

Refactored code to class

Browse Source
This commit is contained in:
Kevin Veen-Birkenbach 2023-09-03 22:16:34 +02:00
parent 0b67080016
commit 276c22f72e
1 changed files with 102 additions and 96 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

192
fetch-all-data.py
View File

@ -2,108 +2,113 @@ import requests
import argparse import argparse
import json import json
def create_headers(api_key): class BaserowAPI:
"""Create headers for API requests.""" def __init__(self, base_url, api_key, verbose=False):
return { self.base_url = base_url
"Authorization": f"Token {api_key}", self.api_key = api_key
"Content-Type": "application/json" self.verbose = verbose
}
def handle_api_response(response, verbose): def create_headers(self):
"""Handle API response, check for errors and decode JSON.""" """Create headers for API requests."""
if verbose: return {
print("[INFO] Handling API response...") "Authorization": f"Token {self.api_key}",
print("Response Status Code:", response.status_code) "Content-Type": "application/json"
print("Response Headers:", response.headers) }
if response.status_code != 200:
print(f"Error: Received status code {response.status_code} from Baserow API.")
print("Response content:", response.content.decode())
return None
try: def handle_api_response(self, response):
return response.json() """Handle API response, check for errors and decode JSON."""
except requests.RequestsJSONDecodeError: if self.verbose:
print("Error: Failed to decode the response as JSON.") print("[INFO] Handling API response...")
return None print("Response Status Code:", response.status_code)
print("Response Headers:", response.headers)
if response.status_code != 200:
print(f"Error: Received status code {response.status_code} from Baserow API.")
print("Response content:", response.content.decode())
return None
def get_all_rows_from_table(base_url, api_key, table_id, verbose): try:
if verbose: return response.json()
print(f"[INFO] Fetching all rows from table with ID: {table_id}...") except requests.RequestsJSONDecodeError:
headers = create_headers(api_key) print("Error: Failed to decode the response as JSON.")
rows = [] return None
next_url = f"{base_url}database/rows/table/{table_id}/"
while next_url: def get_all_rows_from_table(self, table_id):
response = requests.get(next_url, headers=headers) if self.verbose:
if verbose: print(f"[INFO] Fetching all rows from table with ID: {table_id}...")
headers = self.create_headers()
rows = []
next_url = f"{self.base_url}database/rows/table/{table_id}/"
while next_url:
response = requests.get(next_url, headers=headers)
if self.verbose:
print("Headers:", headers)
print("Requesting:", next_url)
data = self.handle_api_response(response)
if not data:
break
rows.extend(data['results'])
next_url = data['next']
return rows
def get_all_tables_from_database(self, database_id):
if self.verbose:
print(f"[INFO] Fetching all tables from database with ID: {database_id}...")
headers = self.create_headers()
response = requests.get(f"{self.base_url}database/tables/database/{database_id}/", headers=headers)
if self.verbose:
print("Headers:", headers) print("Headers:", headers)
print("Requesting:", next_url) return self.handle_api_response(response) or []
data = handle_api_response(response, verbose)
if not data:
break
rows.extend(data['results'])
next_url = data['next']
return rows def get_all_data_from_database(self, database_id):
if self.verbose:
print(f"[INFO] Fetching all data from database with ID: {database_id}...")
tables = self.get_all_tables_from_database(database_id)
data = {}
def get_all_tables_from_database(base_url, api_key, database_id, verbose): for table in tables:
if verbose: table_id = table['id']
print(f"[INFO] Fetching all tables from database with ID: {database_id}...") table_name = table['name']
headers = create_headers(api_key) data[table_name] = self.get_all_rows_from_table(table_id)
response = requests.get(f"{base_url}database/tables/database/{database_id}/", headers=headers)
if verbose:
print("Headers:", headers)
return handle_api_response(response, verbose) or []
def get_all_data_from_database(base_url, api_key, database_id, verbose): return data
if verbose:
print(f"[INFO] Fetching all data from database with ID: {database_id}...")
tables = get_all_tables_from_database(base_url, api_key, database_id, verbose)
data = {}
for table in tables: def fetch_fields_for_table(self, table_id):
table_id = table['id'] """Fetch fields for a given table."""
table_name = table['name'] headers = self.create_headers()
data[table_name] = get_all_rows_from_table(base_url, api_key, table_id, verbose) response = requests.get(f"{self.base_url}database/fields/table/{table_id}/", headers=headers)
if response.status_code == 200:
return response.json()
else:
raise Exception(f"Failed to fetch fields for table {table_id}. Status code: {response.status_code}")
return data def merge_tables_on_reference(self, tables_data):
if self.verbose:
print(
"Merge tables based on references.\n"
"Assumes that a reference from one table to another is represented by a field in the dictionary\n"
"that has the same name as the referenced table and contains the ID of the referenced row.\n"
)
# Create a mapping of table names to their rows indexed by ID
indexed_data = {table_name: {row['id']: row for row in rows} for table_name, rows in tables_data.items()}
def fetch_fields_for_table(base_url, api_key, table_id): # Fetch field information for each table and identify link fields
"""Fetch fields for a given table.""" link_fields = {}
headers = create_headers(api_key) for table_name in tables_data:
response = requests.get(f"{base_url}database/fields/table/{table_id}/", headers=headers) fields = self.fetch_fields_for_table(table_name)
if response.status_code == 200: link_fields[table_name] = [field for field in fields if field['type'] == 'link_row']
return response.json()
else:
raise Exception(f"Failed to fetch fields for table {table_id}. Status code: {response.status_code}")
def merge_tables_on_reference(base_url, api_key, tables_data,verbose):
if verbose:
print(
"Merge tables based on references.\n"
"Assumes that a reference from one table to another is represented by a field in the dictionary\n"
"that has the same name as the referenced table and contains the ID of the referenced row.\n"
)
# Create a mapping of table names to their rows indexed by ID
indexed_data = {table_name: {row['id']: row for row in rows} for table_name, rows in tables_data.items()}
# Fetch field information for each table and identify link fields
link_fields = {}
for table_name in tables_data:
fields = fetch_fields_for_table(base_url, api_key,table_name)
link_fields[table_name] = [field for field in fields if field['type'] == 'link_row']
# Embed referenced data into tables
for table_name, rows in tables_data.items():
for row in rows:
for link_field in link_fields[table_name]:
field_name = link_field['name']
referenced_table_id = link_field['link_row_table_id']
if field_name in row and row[field_name] in indexed_data[referenced_table_id]:
if verbose: print("Embed the referenced row data under the reference field")
row[field_name] = indexed_data[referenced_table_id][row[field_name]]
return tables_data
# Embed referenced data into tables
for table_name, rows in tables_data.items():
for row in rows:
for link_field in link_fields[table_name]:
field_name = link_field['name']
referenced_table_id = link_field['link_row_table_id']
if field_name in row and row[field_name] in indexed_data[referenced_table_id]:
if self.verbose: print("Embed the referenced row data under the reference field")
row[field_name] = indexed_data[referenced_table_id][row[field_name]]
return tables_data
if __name__ == "__main__": if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Fetch all data from a Baserow database.") parser = argparse.ArgumentParser(description="Fetch all data from a Baserow database.")
@ -116,6 +121,7 @@ if __name__ == "__main__":
parser.add_argument("--quiet", action="store_true", help="Suppress output of json") parser.add_argument("--quiet", action="store_true", help="Suppress output of json")
args = parser.parse_args() args = parser.parse_args()
api = BaserowAPI(args.base_url, args.api_key, args.verbose)
if not args.database_id and not args.table_ids: if not args.database_id and not args.table_ids:
print("Error: Either database_id or table_ids must be provided.") print("Error: Either database_id or table_ids must be provided.")
@ -125,14 +131,14 @@ if __name__ == "__main__":
table_ids = args.table_ids.split(',') table_ids = args.table_ids.split(',')
tables_data = {} tables_data = {}
for table_id in table_ids: for table_id in table_ids:
table_data = get_all_rows_from_table(args.base_url, args.api_key, table_id.strip(), args.verbose) table_data = api.get_all_rows_from_table(table_id.strip())
tables_data[table_id] = table_data tables_data[table_id] = table_data
if args.matrix: if args.matrix:
merged_data = merge_tables_on_reference(args.base_url, args.api_key,tables_data, args.verbose) merged_data = api.merge_tables_on_reference(tables_data)
if not args.quiet: print(json.dumps(merged_data, indent=4)) if not args.quiet: print(json.dumps(merged_data, indent=4))
else: else:
if not args.quiet: print(json.dumps(tables_data, indent=4)) if not args.quiet: print(json.dumps(tables_data, indent=4))
else: else:
all_data = get_all_data_from_database(args.base_url, args.api_key, args.database_id, args.verbose) all_data = api.get_all_data_from_database(args.database_id)
if not args.quiet: print(json.dumps(all_data, indent=4)) if not args.quiet: print(json.dumps(all_data, indent=4))