[2ff88f42] refactor(connector-superoffice): finalize production readiness cleanup

- Integrated centralized logging system in all modules.
- Extracted all IDs and ProgIds into a separate .
- Refactored  and  for cleaner dependency management.
- Included updated discovery and inspection utilities.
- Verified end-to-end workflow stability.
This commit is contained in:
2026-02-10 12:43:26 +00:00
parent 42f1679377
commit 8e863b259e
9 changed files with 185 additions and 255 deletions

View File

@@ -1,78 +1,49 @@
import os
import logging
import json
from dotenv import load_dotenv
from config import Config
from logging_config import setup_logging
from auth_handler import AuthHandler
from superoffice_client import SuperOfficeClient
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
logger = setup_logging("discovery")
def get_list_items_by_prog_id(client, prog_id, entity_name):
"""Fetches and prints list items for a specific ProgId."""
logger.info(f"--- Fetching list items for {entity_name} ProgId: {prog_id} ---")
# The endpoint for user-defined lists is typically generic
list_url = client._get_url(f"v1/List/UserDefinedField/{prog_id}")
try:
list_resp = client.session.get(list_url, headers=client._get_headers())
list_resp.raise_for_status()
list_items = list_resp.json()
if list_items.get("value"):
print(" --- List Items Found ---")
print(f" --- List Items Found for {prog_id} ---")
for item in list_items["value"]:
print(f" ID: {item.get('Id'):<5} | Name: {item.get('Name')}")
print(" ------------------------")
return {item.get('Name'): item.get('Id') for item in list_items["value"]}
else:
print(" (No list items found or unexpected response structure)")
return None
except Exception as list_e:
logger.error(f" Failed to fetch list items for {prog_id}: {list_e}")
if hasattr(list_e, 'response') and list_e.response is not None:
logger.error(f" List fetch details: {list_e.response.text}")
return None
except Exception as e:
logger.error(f"Failed to fetch list items for {prog_id}: {e}")
return None
def get_activity_types(client):
"""Fetches available activity types."""
logger.info("--- Fetching Activity Types ---")
# Common endpoint for activity types
activity_type_url = client._get_url("v1/ActivityType") # Trying direct ActivityType endpoint
url = client._get_url("v1/ActivityType")
try:
resp = client.session.get(activity_type_url, headers=client._get_headers())
resp = client.session.get(url, headers=client._get_headers())
resp.raise_for_status()
activity_types = resp.json()
if activity_types:
print(" --- Activity Types Found ---")
for atype in activity_types:
print(f" ID: {atype.get('Id'):<5} | Name: {atype.get('Name')}")
print(" ------------------------")
return {atype.get('Name'): atype.get('Id') for atype in activity_types}
else:
print(" (No activity types found or unexpected response structure)")
return None
except Exception as e:
logger.error(f" Failed to fetch activity types: {e}")
if hasattr(e, 'response') and e.response is not None:
logger.error(f" Activity type fetch details: {e.response.text}")
return None
logger.error(f"Failed to fetch activity types: {e}")
return None
def main():
load_dotenv(dotenv_path="../.env")
auth = AuthHandler()
client = SuperOfficeClient(auth)
# --- We know the ProgIds, so we query them directly ---
# ProgId for the "MA Status" list on the Person entity
#person_ma_status_prog_id = "SuperOffice:2" # Keep for future reference
#get_list_items_by_prog_id(client, person_ma_status_prog_id, "Person MA Status")
get_activity_types(client)
if __name__ == "__main__":
main()
main()