Get a database connection for the specified server and database. Args: sid: Server ID did: Database ID (OID) conn_id: Unique connection identifier Returns: Tuple of (manager, connection) objects Raises: DatabaseToolError: If connection fail
(sid: int, did: int, conn_id: str)
| 66 | |
| 67 | |
| 68 | def _get_connection(sid: int, did: int, conn_id: str): |
| 69 | """ |
| 70 | Get a database connection for the specified server and database. |
| 71 | |
| 72 | Args: |
| 73 | sid: Server ID |
| 74 | did: Database ID (OID) |
| 75 | conn_id: Unique connection identifier |
| 76 | |
| 77 | Returns: |
| 78 | Tuple of (manager, connection) objects |
| 79 | |
| 80 | Raises: |
| 81 | DatabaseToolError: If connection fails |
| 82 | """ |
| 83 | try: |
| 84 | driver = get_driver(config.PG_DEFAULT_DRIVER) |
| 85 | manager = driver.connection_manager(sid) |
| 86 | |
| 87 | # Get connection - this will create one if it doesn't exist |
| 88 | conn = manager.connection( |
| 89 | did=did, |
| 90 | conn_id=conn_id, |
| 91 | auto_reconnect=False, # Don't auto-reconnect for LLM queries |
| 92 | use_binary_placeholder=True, |
| 93 | array_to_string=True |
| 94 | ) |
| 95 | |
| 96 | return manager, conn |
| 97 | |
| 98 | except Exception as e: |
| 99 | raise DatabaseToolError( |
| 100 | f"Failed to get connection: {str(e)}", |
| 101 | code="CONNECTION_ERROR" |
| 102 | ) |
| 103 | |
| 104 | |
| 105 | def _connect_readonly( |
no test coverage detected