Get the schema information for a database. Uses pgAdmin's SQL templates for version-aware schema listing. Args: sid: Server ID did: Database ID Returns: Dictionary containing schema information organized by schema name
(sid: int, did: int)
| 396 | |
| 397 | |
| 398 | def get_database_schema(sid: int, did: int) -> dict: |
| 399 | """ |
| 400 | Get the schema information for a database. |
| 401 | |
| 402 | Uses pgAdmin's SQL templates for version-aware schema listing. |
| 403 | |
| 404 | Args: |
| 405 | sid: Server ID |
| 406 | did: Database ID |
| 407 | |
| 408 | Returns: |
| 409 | Dictionary containing schema information organized by schema name |
| 410 | """ |
| 411 | conn_id = f"llm_{secrets.choice(range(1, 9999999))}" |
| 412 | manager = None |
| 413 | |
| 414 | try: |
| 415 | manager, conn = _get_connection(sid, did, conn_id) |
| 416 | status, error = _connect_readonly(manager, conn, conn_id) |
| 417 | if not status: |
| 418 | raise DatabaseToolError(f"Connection failed: {error}", |
| 419 | code="CONNECTION_ERROR") |
| 420 | |
| 421 | # Get server version for template selection |
| 422 | sversion = manager.sversion or 0 |
| 423 | |
| 424 | # Build template path with version - the versioned loader will |
| 425 | # find the appropriate directory (e.g., 15_plus, 14_plus, default) |
| 426 | schema_template_path = compile_template_path( |
| 427 | SCHEMAS_TEMPLATE_PATH, sversion |
| 428 | ) |
| 429 | |
| 430 | # Get list of schemas using the template |
| 431 | schema_sql = render_template( |
| 432 | "/".join([schema_template_path, 'sql', 'nodes.sql']), |
| 433 | show_sysobj=False, |
| 434 | scid=None, |
| 435 | schema_restrictions=None |
| 436 | ) |
| 437 | |
| 438 | # Execute in read-only mode |
| 439 | status, _ = conn.execute_void("BEGIN TRANSACTION READ ONLY") |
| 440 | if not status: |
| 441 | raise DatabaseToolError("Failed to start transaction", |
| 442 | code="TRANSACTION_ERROR") |
| 443 | |
| 444 | try: |
| 445 | status, schema_res = conn.execute_dict(schema_sql) |
| 446 | if not status: |
| 447 | raise DatabaseToolError(f"Schema query failed: {schema_res}", |
| 448 | code="QUERY_ERROR") |
| 449 | |
| 450 | schemas = {} |
| 451 | table_template_path = compile_template_path( |
| 452 | TABLES_TEMPLATE_PATH, sversion |
| 453 | ) |
| 454 | |
| 455 | for schema_row in schema_res.get('rows', []): |
no test coverage detected