agenthink 0.1.33

A unified agent framework for connecting workflows, databases, and agents

DBConnector Library DocumentationThe DBConnector library is a robust Python-based utility designed to manage multi-database connections dynamically. It integrates with Azure Blob Storage for configuration management and Azure Key Vault for secure credential retrieval. The library supports MySQL, Microsoft SQL Server (MSSQL), and PostgreSQL.Table of ContentsCore ArchitectureInitialization & AuthenticationCore MethodsSecurity FeaturesUsage ExampleCore ArchitectureThe library follows a Singleton-like pattern using a class-level cache to ensure that only one instance of the connector exists per session, optimizing resource usage and connection pooling.Key Components:Azure Blob Storage: Used to store a JSON manifest (datastores_output.json) that defines which databases a specific user/workflow has access to.Azure Key Vault: Stores the sensitive credentials (host, port, username, password) for each database.Connection Dictionary: An internal mapping (connection_object_dict) that stores active connection objects keyed by the database name.Initialization & AuthenticationThe get MethodInstead of standard instantiation, use the @classmethod get to retrieve an instance.Pythonconnector = DBConnector.get(session_id="123", user_id="user_01", workflow_id="wf_abc")

Authentication FlowIdentity: Uses ClientSecretCredential (Service Principal) to authenticate with Azure services.Manifest Fetching: Downloads the datastore list from Azure Blob Storage based on the workflow_id and user_id path.Secret Retrieval: Iterates through the manifest, cleanses the "key" to match Azure naming conventions, and fetches the JSON secret from the Key Vault.Automatic Connection: Based on the datastore_type, it automatically initializes the appropriate driver (mysql-connector, pyodbc, or psycopg2).Core Methods1. Database ExplorationMethodDescriptiondisplay_connections()Returns a string representation of all active DB connections.display_tables(db_name)Returns a list of all base tables in the specified database.get_schema(db_name, table_name)Returns column names, data types, and nullability for a specific table.2. Data Operationsexecute_query(db_name, query): Executes a read-only SQL statement. It returns the result set as a list of tuples.insert_data(db_name, table_name, data): Performs an INSERT operation.data: A dictionary where keys are column names and values are the data to insert.get_data(db_name, table_name, num_rows=5): A convenience method to quickly preview the first few rows of a table.Security FeaturesRead-Only EnforcementThe execute_query method contains a strict safeguard. It checks the beginning of every SQL string against a list of approved Read-Only Prefixes:SELECT, WITH, SHOW, DESCRIBE, DESC, EXPLAINIf a query starts with DELETE, DROP, or UPDATE, the method will return None and refuse execution.Secret SanitizationAzure Key Vault has strict naming requirements (alphanumerics and hyphens only). The library includes a __sanitize_secret_name helper that:Converts spaces and underscores to hyphens.Removes any non-compliant special characters.Usage ExamplePython# 1. Initialize the connector db_manager = DBConnector.get("session_45", "john_doe", "marketing_workflow")

2. List available tables in the 'CustomerDB'

tables = db_manager.display_tables("CustomerDB") print(f"Available tables: {tables}")

3. Fetch data safely

query = "SELECT name, email FROM Users WHERE status = 'active'" results = db_manager.execute_query("CustomerDB", query)

4. Insert a log entry

log_data = {"event": "login", "user_id": 101, "status": "success"} db_manager.insert_data("LogDB", "activity_logs", log_data)