Establish a read-only connection. Sets the application_name to identify this as an LLM connection and ensures the connection is in read-only mode. Args: _manager: The server manager (unused, kept for API consistency) conn: The connection object conn_id: Con
(
_manager, conn, conn_id: str
)
| 103 | |
| 104 | |
| 105 | def _connect_readonly( |
| 106 | _manager, conn, conn_id: str |
| 107 | ) -> tuple[bool, Optional[str]]: |
| 108 | """ |
| 109 | Establish a read-only connection. |
| 110 | |
| 111 | Sets the application_name to identify this as an LLM connection |
| 112 | and ensures the connection is in read-only mode. |
| 113 | |
| 114 | Args: |
| 115 | _manager: The server manager (unused, kept for API consistency) |
| 116 | conn: The connection object |
| 117 | conn_id: Connection identifier |
| 118 | |
| 119 | Returns: |
| 120 | Tuple of (success, error_message) |
| 121 | """ |
| 122 | try: |
| 123 | # Connect if not already connected |
| 124 | if not conn.connected(): |
| 125 | status, msg = conn.connect() |
| 126 | if not status: |
| 127 | return False, msg |
| 128 | |
| 129 | # Set application name via SQL - this is thread-safe and doesn't |
| 130 | # require environment variables. The name will be visible in |
| 131 | # pg_stat_activity to identify LLM connections. |
| 132 | app_name = f'{LLM_APP_NAME_PREFIX} - {conn_id}' |
| 133 | # Escape single quotes in the app name for safety |
| 134 | app_name_escaped = app_name.replace("'", "''") |
| 135 | status, _ = conn.execute_void( |
| 136 | f"SET application_name = '{app_name_escaped}'" |
| 137 | ) |
| 138 | if not status: |
| 139 | # Non-fatal - connection still works without custom app name |
| 140 | pass |
| 141 | |
| 142 | return True, None |
| 143 | |
| 144 | except Exception as e: |
| 145 | return False, str(e) |
| 146 | |
| 147 | |
| 148 | def _first_real_keyword(statement) -> str: |
no test coverage detected