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

Refactored class

Browse Source
This commit is contained in:
Kevin Veen-Birkenbach 2023-09-04 18:04:41 +02:00
parent 4caa037ab5
commit a1ec0b9760
1 changed files with 43 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
baserow_api.py
View File

@ -6,6 +6,8 @@ class BaserowAPI:
self.base_url = base_url self.base_url = base_url
self.api_key = api_key self.api_key = api_key
self.verbose = verbose self.verbose = verbose
self.headers = create_headers(self)
self.print_verbose_message("Headers:", headers)
def create_headers(self): def create_headers(self):
"""Create headers for API requests.""" """Create headers for API requests."""
@ -14,12 +16,15 @@ class BaserowAPI:
"Content-Type": "application/json" "Content-Type": "application/json"
} }
def print_verbose_message(self, message):
if self.verbose:
print(message)
def handle_api_response(self, response): def handle_api_response(self, response):
"""Handle API response, check for errors and decode JSON.""" """Handle API response, check for errors and decode JSON."""
if self.verbose: self.print_verbose_message("[INFO] Handling API response...")
print("[INFO] Handling API response...") self.print_verbose_message("Response Status Code:", response.status_code)
print("Response Status Code:", response.status_code) self.print_verbose_message("Response Headers:", response.headers)
print("Response Headers:", response.headers)
if response.status_code != 200: if response.status_code != 200:
print(f"Error: Received status code {response.status_code} from Baserow API.") print(f"Error: Received status code {response.status_code} from Baserow API.")
print("Response content:", response.content.decode()) print("Response content:", response.content.decode())
@ -34,15 +39,12 @@ class BaserowAPI:
def get_all_rows_from_table(self, table_id): def get_all_rows_from_table(self, table_id):
if self.verbose: if self.verbose:
print(f"[INFO] Fetching all rows from table with ID: {table_id}...") print(f"[INFO] Fetching all rows from table with ID: {table_id}...")
headers = self.create_headers()
rows = [] rows = []
next_url = f"{self.base_url}database/rows/table/{table_id}/" next_url = f"{self.base_url}database/rows/table/{table_id}/"
while next_url: while next_url:
response = requests.get(next_url, headers=headers) response = requests.get(next_url, headers=self.headers)
if self.verbose: self.print_verbose_message("Requesting:", next_url)
print("Headers:", headers)
print("Requesting:", next_url)
data = self.handle_api_response(response) data = self.handle_api_response(response)
if not data: if not data:
break break
@ -52,17 +54,12 @@ class BaserowAPI:
return rows return rows
def get_all_tables_from_database(self, database_id): def get_all_tables_from_database(self, database_id):
if self.verbose: self.print_verbose_message("[INFO] Fetching all tables from database with ID: {database_id}...")
print(f"[INFO] Fetching all tables from database with ID: {database_id}...") response = requests.get(f"{self.base_url}database/tables/database/{database_id}/", headers=self.headers)
headers = self.create_headers()
response = requests.get(f"{self.base_url}database/tables/database/{database_id}/", headers=headers)
if self.verbose:
print("Headers:", headers)
return self.handle_api_response(response) or [] return self.handle_api_response(response) or []
def get_all_data_from_database(self, database_id): def get_all_data_from_database(self, database_id):
if self.verbose: self.print_verbose_message("[INFO] Fetching all data from database with ID: {database_id}...")
print(f"[INFO] Fetching all data from database with ID: {database_id}...")
tables = self.get_all_tables_from_database(database_id) tables = self.get_all_tables_from_database(database_id)
data = {} data = {}
@ -75,43 +72,46 @@ class BaserowAPI:
def fetch_fields_for_table(self, table_id): def fetch_fields_for_table(self, table_id):
"""Fetch fields for a given table.""" """Fetch fields for a given table."""
headers = self.create_headers() response = requests.get(f"{self.base_url}database/fields/table/{table_id}/", headers=self.headers)
response = requests.get(f"{self.base_url}database/fields/table/{table_id}/", headers=headers)
if response.status_code == 200: if response.status_code == 200:
return response.json() return response.json()
else:
raise Exception(f"Failed to fetch fields for table {table_id}. Status code: {response.status_code}") raise Exception(f"Failed to fetch fields for table {table_id}. Status code: {response.status_code}")
def merge_tables_on_reference(self, tables_data): def merge_tables_on_reference(self, tables_data):
if self.verbose: self.print_verbose_message("Merging tables based on references...")
print("Merging tables based on references...") indexed_data = self.index_tables_by_id(tables_data)
link_fields = self.get_link_fields_for_all_tables(tables_data)
self.embed_referenced_data_into_tables(tables_data, indexed_data, link_fields)
self.print_verbose_message("Merged Tables Data:", tables_data)
return tables_data
# Create a mapping of table names to their rows indexed by ID def index_tables_by_id(self, tables_data):
indexed_data = {table_name: {row['id']: row for row in rows} for table_name, rows in tables_data.items()} indexed_data = {table_name: {row['id']: row for row in rows} for table_name, rows in tables_data.items()}
if self.verbose: self.print_verbose_message("Indexed Data: {indexed_data}")
print("Indexed Data:", indexed_data) return indexed_data
# Fetch field information for each table and identify link fields def get_link_fields_for_all_tables(self, tables_data):
link_fields = {} link_fields = {}
for table_name in tables_data: for table_name in tables_data:
fields = self.fetch_fields_for_table(table_name) link_fields_for_table = self.get_link_fields_for_table(table_name)
link_fields_for_table = [field for field in fields if field['type'] == 'link_row']
link_fields[table_name] = link_fields_for_table link_fields[table_name] = link_fields_for_table
if self.verbose: self.print_verbose_message("Link Fields For Table: {link_fields_for_table}")
print("Link Fields:", link_fields) self.print_verbose_message("Link Fields: {link_fields}")
print("Link Fields For Table:", link_fields_for_table) return link_fields
# Embed referenced data into tables def get_link_fields_for_table(self, table_name):
fields = self.fetch_fields_for_table(table_name)
return [field for field in fields if field['type'] == 'link_row']
def embed_referenced_data_into_tables(self, tables_data, indexed_data, link_fields):
for table_name, rows in tables_data.items(): for table_name, rows in tables_data.items():
for row in rows: for row in rows:
self.embed_data_for_row(row, table_name, indexed_data, link_fields)
def embed_data_for_row(self, row, table_name, indexed_data, link_fields):
for link_field in link_fields[table_name]: for link_field in link_fields[table_name]:
field_name = link_field['name'] field_name = link_field['name']
referenced_table_id = link_field['link_row_table_id'] 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 field_name in row and row[field_name] in indexed_data[referenced_table_id]:
if self.verbose: self.print_verbose_message(f"Embedding referenced data for field {field_name} in table {table_name}")
print(f"Embedding referenced data for field {field_name} in table {table_name}")
row[field_name] = indexed_data[referenced_table_id][row[field_name]] row[field_name] = indexed_data[referenced_table_id][row[field_name]]
if self.verbose:
print("Merged Tables Data:", tables_data)
return tables_data