From 7d32fd403620eeffd59c9902c21148932d56e86e Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 8 Apr 2026 11:58:38 +0800 Subject: [PATCH 1/4] feat(introspection): add PostgreSQL server status overview support Add comprehensive server status introspection for PostgreSQL backend: - SyncPostgreSQLStatusIntrospector and AsyncPostgreSQLStatusIntrospector - Support for configuration, performance, connections, storage info - PostgreSQL-specific: WAL, replication, archive, security, extensions - CLI 'status' subcommand with JSON and Rich output formats - Lazy initialization via introspector.status property --- .../backend/impl/postgres/__main__.py | 469 +++++- .../impl/postgres/introspection/__init__.py | 6 + .../postgres/introspection/introspector.py | 20 + .../introspection/status_introspector.py | 1292 +++++++++++++++++ 4 files changed, 1786 insertions(+), 1 deletion(-) create mode 100644 src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py diff --git a/src/rhosocial/activerecord/backend/impl/postgres/__main__.py b/src/rhosocial/activerecord/backend/impl/postgres/__main__.py index a0843e25..d02c674f 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/__main__.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/__main__.py @@ -48,6 +48,7 @@ ) from rhosocial.activerecord.backend.errors import ConnectionError, QueryError from rhosocial.activerecord.backend.output import JsonOutputProvider, CsvOutputProvider, TsvOutputProvider +from rhosocial.activerecord.backend.introspection.status import StatusCategory from rhosocial.activerecord.backend.dialect.protocols import ( WindowFunctionSupport, CTESupport, @@ -100,6 +101,8 @@ "indexes", "foreign-keys", "triggers", "database" ] +STATUS_TYPES = ["all", "config", "performance", "connections", "storage", "databases", "users"] + def _serialize_for_output(obj: Any) -> Any: """Serialize object for JSON output, handling non-serializable types. @@ -357,6 +360,50 @@ def parse_args(): help="Include system tables in output", ) + # status subcommand + status_parser = subparsers.add_parser( + "status", + help="Display server status overview", + parents=[parent_parser], + epilog="""Examples: + # Show complete status overview + %(prog)s all --database mydb + + # Show configuration parameters only + %(prog)s config --database mydb + + # Show performance metrics only + %(prog)s performance --database mydb + + # Show connection information + %(prog)s connections --database mydb + + # Show storage information + %(prog)s storage --database mydb + + # Show databases list + %(prog)s databases --database mydb + + # Show users list + %(prog)s users --database mydb + + # Output as JSON + %(prog)s all --database mydb -o json + + # Using environment variables for connection + export POSTGRES_HOST=localhost POSTGRES_DATABASE=mydb POSTGRES_USER=postgres POSTGRES_PASSWORD=secret + %(prog)s all +""", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + status_parser.add_argument( + "type", + nargs="?", + default="all", + choices=STATUS_TYPES, + help="Status type: all (default), config, performance, connections, storage, databases, users", + ) + return parser.parse_args() @@ -925,12 +972,410 @@ async def handle_introspect_async(args, backend: AsyncPostgresBackend, provider: await backend.disconnect() +def handle_status_sync(args, backend: PostgresBackend, provider: Any): + """Handle status subcommand synchronously.""" + try: + backend.connect() + backend.introspect_and_adapt() + + status_introspector = backend.introspector.status + status_type = args.type + + if status_type == "all": + # Get complete overview + status = status_introspector.get_overview() + if args.output == "json" or not RICH_AVAILABLE: + print(json.dumps(_serialize_for_output(status), indent=2)) + else: + _display_status_rich(status, args.verbose) + elif status_type == "config": + config_items = status_introspector.list_configuration(StatusCategory.CONFIGURATION) + data = _serialize_for_output(config_items) + provider.display_results(data, title="Configuration") + elif status_type == "performance": + perf_items = status_introspector.list_configuration(StatusCategory.PERFORMANCE) + data = _serialize_for_output(perf_items) + provider.display_results(data, title="Performance") + elif status_type == "connections": + conn_info = status_introspector.get_connection_info() + data = _serialize_for_output(conn_info) + provider.display_results([data], title="Connections") + elif status_type == "storage": + storage_info = status_introspector.get_storage_info() + data = _serialize_for_output(storage_info) + provider.display_results([data], title="Storage") + elif status_type == "databases": + databases = status_introspector.list_databases() + data = _serialize_for_output(databases) + provider.display_results(data, title="Databases") + elif status_type == "users": + users = status_introspector.list_users() + data = _serialize_for_output(users) + provider.display_results(data, title="Users") + + except ConnectionError as e: + provider.display_connection_error(e) + sys.exit(1) + except QueryError as e: + provider.display_query_error(e) + sys.exit(1) + except Exception as e: + print(f"Error during status introspection: {e}", file=sys.stderr) + sys.exit(1) + finally: + if backend._connection: # type: ignore + backend.disconnect() + + +async def handle_status_async(args, backend: AsyncPostgresBackend, provider: Any): + """Handle status subcommand asynchronously.""" + try: + await backend.connect() + await backend.introspect_and_adapt() + + status_introspector = backend.introspector.status + status_type = args.type + + if status_type == "all": + # Get complete overview + status = await status_introspector.get_overview() + if args.output == "json" or not RICH_AVAILABLE: + print(json.dumps(_serialize_for_output(status), indent=2)) + else: + _display_status_rich(status, args.verbose) + elif status_type == "config": + config_items = await status_introspector.list_configuration(StatusCategory.CONFIGURATION) + data = _serialize_for_output(config_items) + provider.display_results(data, title="Configuration") + elif status_type == "performance": + perf_items = await status_introspector.list_configuration(StatusCategory.PERFORMANCE) + data = _serialize_for_output(perf_items) + provider.display_results(data, title="Performance") + elif status_type == "connections": + conn_info = await status_introspector.get_connection_info() + data = _serialize_for_output(conn_info) + provider.display_results([data], title="Connections") + elif status_type == "storage": + storage_info = await status_introspector.get_storage_info() + data = _serialize_for_output(storage_info) + provider.display_results([data], title="Storage") + elif status_type == "databases": + databases = await status_introspector.list_databases() + data = _serialize_for_output(databases) + provider.display_results(data, title="Databases") + elif status_type == "users": + users = await status_introspector.list_users() + data = _serialize_for_output(users) + provider.display_results(data, title="Users") + + except ConnectionError as e: + provider.display_connection_error(e) + sys.exit(1) + except QueryError as e: + provider.display_query_error(e) + sys.exit(1) + except Exception as e: + print(f"Error during status introspection: {e}", file=sys.stderr) + sys.exit(1) + finally: + if backend._connection: # type: ignore + await backend.disconnect() + + +def _display_status_rich(status: Any, verbose: int = 0): + """Display status using rich console. + + Args: + status: ServerOverview object + verbose: Verbosity level for output detail + """ + from rich.console import Console + from rich.table import Table + + console = Console(force_terminal=True) + + # Header + console.print("\n[bold cyan]PostgreSQL Server Status[/bold cyan]\n") + console.print(f"[bold]Version:[/bold] {status.server_version}") + console.print(f"[bold]Vendor:[/bold] {status.server_vendor}") + + # Session info + if hasattr(status, 'session') and status.session: + session = status.session + console.print() + console.print("[bold green]Session[/bold green]") + if session.user: + console.print(f" [bold]User:[/bold] {session.user}") + if session.database: + console.print(f" [bold]Database:[/bold] {session.database}") + if session.schema: + console.print(f" [bold]Schema:[/bold] {session.schema}") + if session.host: + console.print(f" [bold]Host:[/bold] {session.host}") + if session.ssl_enabled is not None: + ssl_status = "Enabled" if session.ssl_enabled else "Disabled" + console.print(f" [bold]SSL:[/bold] {ssl_status}") + if session.ssl_enabled and session.ssl_version: + console.print(f" [bold]SSL Version:[/bold] {session.ssl_version}") + if session.ssl_enabled and session.ssl_cipher: + console.print(f" [bold]SSL Cipher:[/bold] {session.ssl_cipher}") + if session.password_used is not None: + auth_method = "Password" if session.password_used else "Other" + console.print(f" [bold]Auth Method:[/bold] {auth_method}") + + # Connection info + if hasattr(status, 'connections') and status.connections: + conn = status.connections + if conn.active_count is not None: + console.print(f"[bold]Active Connections:[/bold] {conn.active_count}") + if conn.max_connections is not None: + console.print(f"[bold]Max Connections:[/bold] {conn.max_connections}") + + console.print() + + # Configuration section + config_items = [item for item in status.configuration + if item.category == StatusCategory.CONFIGURATION] + if config_items: + console.print("[bold green]Configuration[/bold green]") + config_table = Table(show_header=True, header_style="bold") + config_table.add_column("Parameter") + config_table.add_column("Value") + if verbose >= 1: + config_table.add_column("Description") + config_table.add_column("Readonly") + + for item in config_items: + row = [item.name, str(item.value)] + if verbose >= 1: + row.extend([ + item.description or "", + "Yes" if item.is_readonly else "No" + ]) + config_table.add_row(*row) + + console.print(config_table) + console.print() + + # Performance section + perf_items = [item for item in status.configuration + if item.category == StatusCategory.PERFORMANCE] + if perf_items: + console.print("[bold green]Performance[/bold green]") + perf_table = Table(show_header=True, header_style="bold") + perf_table.add_column("Parameter") + perf_table.add_column("Value") + if verbose >= 1: + perf_table.add_column("Unit") + + for item in perf_items: + row = [item.name, str(item.value)] + if verbose >= 1: + row.append(item.unit or "") + perf_table.add_row(*row) + + console.print(perf_table) + console.print() + + # Storage section + if hasattr(status, 'storage') and status.storage: + console.print("[bold green]Storage[/bold green]") + storage_table = Table(show_header=True, header_style="bold") + storage_table.add_column("Metric") + storage_table.add_column("Value") + + if status.storage.total_size_bytes is not None: + storage_table.add_row("Total Size", _format_size(status.storage.total_size_bytes)) + + console.print(storage_table) + console.print() + + # Databases section + if status.databases: + console.print("[bold green]Databases[/bold green]") + db_table = Table(show_header=True, header_style="bold") + db_table.add_column("Name") + db_table.add_column("Size") + if verbose >= 1: + db_table.add_column("Tables") + db_table.add_column("Views") + + for db in status.databases: + row = [db.name, _format_size(db.size_bytes) if db.size_bytes else "N/A"] + if verbose >= 1: + row.append(str(db.table_count) if db.table_count is not None else "N/A") + row.append(str(db.view_count) if db.view_count is not None else "N/A") + db_table.add_row(*row) + + console.print(db_table) + console.print() + + # Show schema details for current database + for db in status.databases: + if db.extra and 'schemas' in db.extra: + schemas = db.extra['schemas'] + if schemas: + console.print(f"[bold green]Schemas in {db.name}[/bold green]") + schema_table = Table(show_header=True, header_style="bold") + schema_table.add_column("Schema") + schema_table.add_column("Tables") + schema_table.add_column("Views") + + for schema_name, counts in sorted(schemas.items()): + schema_table.add_row( + schema_name, + str(counts.get('tables', 0)), + str(counts.get('views', 0)) + ) + + console.print(schema_table) + console.print() + + # Users section + if status.users: + console.print("[bold green]Users[/bold green]") + user_table = Table(show_header=True, header_style="bold") + user_table.add_column("Name") + user_table.add_column("Superuser") + + for user in status.users: + user_table.add_row(user.name, "Yes" if user.is_superuser else "No") + + console.print(user_table) + console.print() + + # WAL section + if hasattr(status, 'wal') and status.wal: + wal = status.wal + console.print("[bold green]WAL (Write-Ahead Logging)[/bold green]") + wal_table = Table(show_header=True, header_style="bold") + wal_table.add_column("Parameter") + wal_table.add_column("Value") + + if wal.wal_level: + wal_table.add_row("WAL Level", str(wal.wal_level)) + if wal.wal_files is not None: + wal_table.add_row("WAL Files", str(wal.wal_files)) + if wal.wal_size_bytes is not None: + wal_table.add_row("WAL Size", _format_size(wal.wal_size_bytes)) + if wal.checkpoint_count is not None: + wal_table.add_row("Checkpoints", str(wal.checkpoint_count)) + if wal.checkpoint_time: + wal_table.add_row("Last Checkpoint", str(wal.checkpoint_time)) + if wal.wal_segments is not None: + wal_table.add_row("WAL Segments", str(wal.wal_segments)) + + console.print(wal_table) + console.print() + + # Replication section + if hasattr(status, 'replication') and status.replication: + repl = status.replication + console.print("[bold green]Replication[/bold green]") + repl_table = Table(show_header=True, header_style="bold") + repl_table.add_column("Parameter") + repl_table.add_column("Value") + + if repl.is_primary is not None: + repl_table.add_row("Role", "Primary" if repl.is_primary else "Standby") + if repl.replication_slots is not None: + repl_table.add_row("Replication Slots", str(repl.replication_slots)) + if repl.active_replications is not None: + repl_table.add_row("Active Replications", str(repl.active_replications)) + if repl.streaming is not None: + repl_table.add_row("Streaming", "Yes" if repl.streaming else "No") + if repl.lag_bytes is not None: + repl_table.add_row("Replication Lag", _format_size(repl.lag_bytes)) + + console.print(repl_table) + console.print() + + # Archive section + if hasattr(status, 'archive') and status.archive: + archive = status.archive + console.print("[bold green]Archive[/bold green]") + archive_table = Table(show_header=True, header_style="bold") + archive_table.add_column("Parameter") + archive_table.add_column("Value") + + if archive.archive_mode: + archive_table.add_row("Archive Mode", str(archive.archive_mode)) + if archive.archive_command: + # Truncate long commands + cmd = str(archive.archive_command) + if len(cmd) > 50: + cmd = cmd[:47] + "..." + archive_table.add_row("Archive Command", cmd) + if archive.archive_timeout is not None: + archive_table.add_row("Archive Timeout", f"{archive.archive_timeout}s") + if archive.archived_count is not None: + archive_table.add_row("Archived Count", str(archive.archived_count)) + if archive.failed_count is not None: + archive_table.add_row("Failed Archives", str(archive.failed_count)) + + console.print(archive_table) + console.print() + + # Security section + if hasattr(status, 'security') and status.security: + security = status.security + console.print("[bold green]Security[/bold green]") + security_table = Table(show_header=True, header_style="bold") + security_table.add_column("Parameter") + security_table.add_column("Value") + + if security.ssl_enabled is not None: + security_table.add_row("SSL Enabled", "Yes" if security.ssl_enabled else "No") + if security.ssl_cert_file: + security_table.add_row("SSL Cert File", str(security.ssl_cert_file)) + if security.ssl_key_file: + security_table.add_row("SSL Key File", str(security.ssl_key_file)) + if security.ssl_ca_file: + security_table.add_row("SSL CA File", str(security.ssl_ca_file)) + if security.password_encryption: + security_table.add_row("Password Encryption", str(security.password_encryption)) + if security.row_security is not None: + security_table.add_row("Row-Level Security", "Enabled" if security.row_security else "Disabled") + + console.print(security_table) + console.print() + + # Extensions section + if hasattr(status, 'extensions') and status.extensions: + console.print("[bold green]Extensions[/bold green]") + ext_table = Table(show_header=True, header_style="bold") + ext_table.add_column("Name") + ext_table.add_column("Version") + ext_table.add_column("Schema") + + for ext in status.extensions: + ext_table.add_row( + ext.name, + ext.version or "N/A", + ext.schema or "N/A" + ) + + console.print(ext_table) + console.print() + + +def _format_size(size_bytes: int) -> str: + """Format byte size to human-readable string.""" + units = ["B", "KB", "MB", "GB", "TB"] + size = float(size_bytes) + for unit in units: + if size < 1024: + return f"{size:.1f} {unit}" + size /= 1024 + return f"{size:.1f} PB" + + def main(): args = parse_args() # Must specify a subcommand if args.command is None: - print("Error: Please specify a command: 'info', 'query', or 'introspect'", file=sys.stderr) + print("Error: Please specify a command: 'info', 'query', 'introspect', or 'status'", file=sys.stderr) print("Use --help for more information.", file=sys.stderr) sys.exit(1) @@ -963,6 +1408,28 @@ def main(): handle_introspect_sync(args, backend, provider) return + # Handle status subcommand + if args.command == "status": + if not args.database: + print("Error: --database is required for status", file=sys.stderr) + sys.exit(1) + + config = PostgresConnectionConfig( + host=args.host, + port=args.port, + database=args.database, + username=args.user, + password=args.password, + ) + + if args.use_async: + backend = AsyncPostgresBackend(connection_config=config) + asyncio.run(handle_status_async(args, backend, provider)) + else: + backend = PostgresBackend(connection_config=config) + handle_status_sync(args, backend, provider) + return + # Handle query subcommand numeric_level = getattr(logging, args.log_level.upper(), None) if not isinstance(numeric_level, int): diff --git a/src/rhosocial/activerecord/backend/impl/postgres/introspection/__init__.py b/src/rhosocial/activerecord/backend/impl/postgres/introspection/__init__.py index 3ee6cc6b..94018285 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/introspection/__init__.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/introspection/__init__.py @@ -3,8 +3,14 @@ SyncPostgreSQLIntrospector, AsyncPostgreSQLIntrospector, ) +from .status_introspector import ( + SyncPostgreSQLStatusIntrospector, + AsyncPostgreSQLStatusIntrospector, +) __all__ = [ "SyncPostgreSQLIntrospector", "AsyncPostgreSQLIntrospector", + "SyncPostgreSQLStatusIntrospector", + "AsyncPostgreSQLStatusIntrospector", ] diff --git a/src/rhosocial/activerecord/backend/impl/postgres/introspection/introspector.py b/src/rhosocial/activerecord/backend/impl/postgres/introspection/introspector.py index 1601221b..4d34772d 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/introspection/introspector.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/introspection/introspector.py @@ -51,6 +51,10 @@ TriggerInfo, ViewInfo, ) +from .status_introspector import ( + SyncPostgreSQLStatusIntrospector, + AsyncPostgreSQLStatusIntrospector, +) class PostgreSQLIntrospectorMixin(IntrospectorMixin): @@ -314,6 +318,14 @@ class SyncPostgreSQLIntrospector(PostgreSQLIntrospectorMixin, SyncAbstractIntros def __init__(self, backend: Any, executor: SyncIntrospectorExecutor) -> None: super().__init__(backend, executor) + self._status_instance: Optional[SyncPostgreSQLStatusIntrospector] = None + + @property + def status(self) -> SyncPostgreSQLStatusIntrospector: + """PostgreSQL status introspector (lazily created).""" + if self._status_instance is None: + self._status_instance = SyncPostgreSQLStatusIntrospector(self._backend) + return self._status_instance class AsyncPostgreSQLIntrospector(PostgreSQLIntrospectorMixin, AsyncAbstractIntrospector): @@ -334,3 +346,11 @@ class AsyncPostgreSQLIntrospector(PostgreSQLIntrospectorMixin, AsyncAbstractIntr def __init__(self, backend: Any, executor: AsyncIntrospectorExecutor) -> None: super().__init__(backend, executor) + self._status_instance: Optional[AsyncPostgreSQLStatusIntrospector] = None + + @property + def status(self) -> AsyncPostgreSQLStatusIntrospector: + """PostgreSQL status introspector (lazily created).""" + if self._status_instance is None: + self._status_instance = AsyncPostgreSQLStatusIntrospector(self._backend) + return self._status_instance diff --git a/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py b/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py new file mode 100644 index 00000000..7d4d2845 --- /dev/null +++ b/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py @@ -0,0 +1,1292 @@ +# src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py +""" +PostgreSQL server status introspector. + +Provides server status information by querying PostgreSQL's pg_settings, +pg_stat_activity, and other system views. + +Design principle: Sync and Async are separate and cannot coexist. +- SyncPostgreSQLStatusIntrospector: for synchronous backends +- AsyncPostgreSQLStatusIntrospector: for asynchronous backends +""" + +from typing import Any, Dict, List, Optional + +from rhosocial.activerecord.backend.introspection.status import ( + StatusItem, + StatusCategory, + ServerOverview, + DatabaseBriefInfo, + UserInfo, + ConnectionInfo, + StorageInfo, + SessionInfo, + WALInfo, + ReplicationInfo, + ArchiveInfo, + SecurityInfo, + ExtensionInfo, + SyncAbstractStatusIntrospector, + AsyncAbstractStatusIntrospector, +) + + +# PostgreSQL settings to include in status overview +# Format: (setting_name, category, description, unit, is_readonly) +POSTGRES_CONFIG_VARIABLES = [ + # Configuration + ("listen_addresses", StatusCategory.CONFIGURATION, "Server listen addresses", None, False), + ("port", StatusCategory.CONFIGURATION, "Server port", None, True), + ("max_connections", StatusCategory.CONFIGURATION, "Maximum connections", None, True), + ("shared_buffers", StatusCategory.CONFIGURATION, "Shared buffers size", "bytes", True), + ("work_mem", StatusCategory.CONFIGURATION, "Work memory size", "bytes", True), + ("maintenance_work_mem", StatusCategory.CONFIGURATION, "Maintenance work memory", "bytes", True), + ("dynamic_shared_memory", StatusCategory.CONFIGURATION, "Dynamic shared memory", None, False), + ("wal_level", StatusCategory.CONFIGURATION, "WAL level", None, False), + ("wal_keep_segments", StatusCategory.CONFIGURATION, "WAL segments", None, False), + ("synchronous_commit", StatusCategory.CONFIGURATION, "Synchronous commit", None, False), + ("checkpoint_timeout", StatusCategory.PERFORMANCE, "Checkpoint timeout", "seconds", False), + ("checkpoint_completion_target", StatusCategory.PERFORMANCE, "Checkpoint completion target", "seconds", False), + ("archive_timeout", StatusCategory.PERFORMANCE, "Archive timeout", "seconds", False), + ("archive_command", StatusCategory.PERFORMANCE, "Archive command", None, False), + ("log_timezone", StatusCategory.CONFIGURATION, "Log timezone", None, False), + ("log_destination", StatusCategory.CONFIGURATION, "Log destination", None, False), + ("logging_collector", StatusCategory.CONFIGURATION, "Logging collector", None, False), + ("cluster_name", StatusCategory.CONFIGURATION, "Cluster name", None, False), + + # Security + ("authentication_timeout", StatusCategory.SECURITY, "Authentication timeout", "seconds", False), + ("password_encryption", StatusCategory.SECURITY, "Password encryption", None, True), + ("ssl", StatusCategory.SECURITY, "SSL enabled", None, False), + ("krb_server_keyfile", StatusCategory.SECURITY, "Kerberos keyfile", None, False), + + # Replication + ("max_wal_senders", StatusCategory.REPLICATION, "Maximum WAL senders", None, False), + ("max_replication_slots", StatusCategory.REPLICATION, "Maximum replication slots", None, False), + ("hot_standby", StatusCategory.REPLICATION, "Hot standby mode", None, False), +] + + +class PostgreSQLStatusIntrospectorMixin: + """Mixin providing shared PostgreSQL status introspection logic. + + Both SyncPostgreSQLStatusIntrospector and AsyncPostgreSQLStatusIntrospector + inherit from this mixin to share: + - Query execution helpers + - _parse_* implementations + """ + + def _get_vendor_name(self) -> str: + """Get PostgreSQL vendor name.""" + return "PostgreSQL" + + def _parse_setting_value(self, value: Any) -> Any: + """Parse PostgreSQL setting value to appropriate Python type. + + PostgreSQL settings can be: + - bool: 'on', 'off', 'true', 'false', 'yes', 'no' + - int: numeric values + - str: string values + - bytes: bytes values with unit + """ + if isinstance(value, bytes): + return value.decode('utf-8') + if isinstance(value, bool): + return value + if isinstance(value, str): + if value in ('on', 'true', 'yes'): + return True + if value in ('off', 'false', 'no'): + return False + if isinstance(value, int): + return value + return str(value) + + def _get_version_string(self) -> str: + """Get PostgreSQL version string.""" + # Try to get version from backend's _server_version_cache attribute first + # (set by introspect_and_adapt() during backend initialization) + version = getattr(self._backend, '_server_version_cache', None) + if version: + return ".".join(str(v) for v in version) + + # Fall back to _version attribute + version = getattr(self._backend, '_version', None) + if version: + return ".".join(str(v) for v in version) + + # Try dialect version + if hasattr(self._backend, '_dialect') and hasattr(self._backend._dialect, 'version'): + version = self._backend._dialect.version + if version: + return ".".join(str(v) for v in version) + + # Last resort: query server directly + try: + version = self._backend.get_server_version() + if version: + return ".".join(str(v) for v in version) + except Exception: + pass + + return "Unknown" + + def _create_status_item( + self, + name: str, + value: Any, + category: StatusCategory, + description: Optional[str] = None, + unit: Optional[str] = None, + is_readonly: bool = False, + ) -> StatusItem: + """Create a StatusItem with parsed value.""" + return StatusItem( + name=name, + value=self._parse_setting_value(value), + category=category, + description=description, + unit=unit, + is_readonly=is_readonly, + ) + + def _build_server_overview( + self, + configuration: List[StatusItem], + performance: List[StatusItem], + connections: ConnectionInfo, + storage: StorageInfo, + databases: List[DatabaseBriefInfo], + users: List[UserInfo], + session: Optional[SessionInfo] = None, + wal: Optional[WALInfo] = None, + replication: Optional[ReplicationInfo] = None, + archive: Optional[ArchiveInfo] = None, + security: Optional[SecurityInfo] = None, + extensions: Optional[List[ExtensionInfo]] = None, + ) -> ServerOverview: + """Build ServerOverview from collected data.""" + return ServerOverview( + server_version=self._get_version_string(), + server_vendor=self._get_vendor_name(), + session=session, + configuration=configuration, + performance=performance, + connections=connections, + storage=storage, + databases=databases, + users=users, + wal=wal, + replication=replication, + archive=archive, + security=security, + extensions=extensions or [], + ) + + +class SyncPostgreSQLStatusIntrospector( + PostgreSQLStatusIntrospectorMixin, SyncAbstractStatusIntrospector +): + """Synchronous PostgreSQL status introspector. + + Usage:: + + backend = PostgreSQLBackend(connection_config=config) + backend.connect() + status = backend.introspector.status.get_overview() + print(status.server_version) + """ + + def __init__(self, backend: Any) -> None: + super().__init__(backend) + + def _exec_query(self, sql: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]: + """Execute a SQL query synchronously.""" + return self._backend.execute(sql, params).data or [] + + def get_overview(self) -> ServerOverview: + """Get complete PostgreSQL status overview.""" + configuration = self.list_configuration() + performance = self.list_performance_metrics() + connections = self.get_connection_info() + storage = self.get_storage_info() + databases = self.list_databases() + users = self.list_users() + session = self.get_session_info() + wal = self.get_wal_info() + replication = self.get_replication_info() + archive = self.get_archive_info() + security = self.get_security_info() + extensions = self.list_extensions() + + return self._build_server_overview( + configuration=configuration, + performance=performance, + connections=connections, + storage=storage, + databases=databases, + users=users, + session=session, + wal=wal, + replication=replication, + archive=archive, + security=security, + extensions=extensions, + ) + + def list_configuration( + self, category: Optional[StatusCategory] = None + ) -> List[StatusItem]: + """List PostgreSQL configuration parameters from pg_settings.""" + items = [] + + # Query pg_settings + sql = "SELECT name, setting, unit FROM pg_settings ORDER BY name" + rows = self._exec_query(sql) + + settings_dict = {row['name']: row for row in rows} + + for name, cat, desc, unit, is_readonly in POSTGRES_CONFIG_VARIABLES: + if category and cat != category: + continue + + # Get setting value from pg_settings + row = settings_dict.get(name) + if row: + item = self._create_status_item( + name=name, + value=row.get('setting'), + category=cat, + description=desc, + unit=row.get('unit') if row.get('unit') else unit, + is_readonly=is_readonly, + ) + items.append(item) + + return items + + def list_performance_metrics( + self, category: Optional[StatusCategory] = None + ) -> List[StatusItem]: + """List PostgreSQL performance metrics from pg_stat_activity.""" + items = [] + + # Query pg_stat_activity for performance-related settings + sql = """ + SELECT name, setting, unit FROM pg_settings + WHERE name IN ('checkpoint_timeout', 'checkpoint_completion_target', 'archive_timeout', 'archive_command') + ORDER BY name + """ + rows = self._exec_query(sql) + + for row in rows: + name = row['name'] + for var_name, var_cat, desc, unit, is_readonly in POSTGRES_CONFIG_VARIABLES: + if name == var_name: + item = self._create_status_item( + name=name, + value=row.get('setting'), + category=var_cat, + description=desc, + unit=row.get('unit') if row.get('unit') else unit, + is_readonly=is_readonly, + ) + items.append(item) + break + + return items + + def get_connection_info(self) -> ConnectionInfo: + """Get connection information from pg_stat_activity.""" + sql = """ + SELECT count(*) as total_connections + FROM pg_stat_activity + WHERE datname = current_database() + """ + rows = self._exec_query(sql) + total = rows[0].get('total_connections', 0) if rows else 0 + + max_conn = None + try: + sql = "SHOW max_connections" + result = self._exec_query(sql) + if result: + max_conn = int(list(result[0].values())[0]) + except Exception: + pass + + return ConnectionInfo( + active_count=total, + max_connections=max_conn, + ) + + def get_storage_info(self) -> StorageInfo: + """Get storage information.""" + db_name = getattr(self._backend.config, 'database', 'postgres') + + # Get database size + sql = "SELECT pg_database_size(%s) as size_bytes" + rows = self._exec_query(sql, (db_name,)) + size_bytes = rows[0].get('size_bytes') if rows else None + + return StorageInfo( + total_size_bytes=size_bytes, + ) + + def list_databases(self) -> List[DatabaseBriefInfo]: + """List databases from pg_database with schema info for current database. + + For the current database, includes schema information with table/view counts + in the extra field. + """ + # Get list of databases + sql = """ + SELECT datname as name, pg_database_size(datname) as size_bytes + FROM pg_database + WHERE datistemplate = false OR datname = current_database() + ORDER BY datname + """ + rows = self._exec_query(sql) + + # Get current database name + current_db = None + try: + result = self._exec_query("SELECT current_database()") + if result: + current_db = result[0].get('current_database') + except Exception: + pass + + # Get schema info for current database + schema_info: Dict[str, Dict[str, int]] = {} + if current_db: + try: + # Query schemata with table/view counts + sql = """ + SELECT + n.nspname as schema_name, + COUNT(CASE WHEN c.relkind = 'r' THEN 1 END) as table_count, + COUNT(CASE WHEN c.relkind = 'v' THEN 1 END) as view_count + FROM pg_namespace n + LEFT JOIN pg_class c ON c.relnamespace = n.oid + WHERE n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast') + GROUP BY n.nspname + ORDER BY n.nspname + """ + schema_rows = self._exec_query(sql) + for row in schema_rows: + schema_name = row.get('schema_name') + if schema_name: + schema_info[schema_name] = { + 'tables': row.get('table_count', 0) or 0, + 'views': row.get('view_count', 0) or 0, + } + except Exception: + pass + + databases = [] + for row in rows: + db_name = row['name'] + db_info = DatabaseBriefInfo( + name=db_name, + size_bytes=row.get('size_bytes'), + ) + + # Add schema info for current database + if db_name == current_db and schema_info: + db_info.extra['schemas'] = schema_info + # Also set total table/view counts + db_info.table_count = sum(s['tables'] for s in schema_info.values()) + db_info.view_count = sum(s['views'] for s in schema_info.values()) + + databases.append(db_info) + + return databases + + def list_users(self) -> List[UserInfo]: + """List users from pg_roles.""" + sql = """ + SELECT rolname as name, rolsuper as is_superuser, + pg_catalog.pg_get_userbyid(oid) as owner + FROM pg_roles + ORDER BY rolname + """ + rows = self._exec_query(sql) + + return [ + UserInfo( + name=row['name'], + is_superuser=row['is_superuser'], + extra={'owner': row.get('owner')} if row.get('owner') else {}, + ) + for row in rows + ] + + def get_session_info(self) -> SessionInfo: + """Get current session/connection information.""" + session = SessionInfo() + + # Get current user + try: + rows = self._exec_query("SELECT current_user") + if rows: + session.user = rows[0].get('current_user') + except Exception: + pass + + # Get current database + try: + rows = self._exec_query("SELECT current_database()") + if rows: + session.database = rows[0].get('current_database') + except Exception: + pass + + # Get current schema + try: + rows = self._exec_query("SELECT current_schema()") + if rows: + session.schema = rows[0].get('current_schema') + except Exception: + pass + + # Get client host address + try: + rows = self._exec_query("SELECT inet_client_addr()") + if rows: + host = rows[0].get('inet_client_addr') + if host: + session.host = str(host) + except Exception: + pass + + # Get SSL status from pg_stat_ssl + try: + rows = self._exec_query(""" + SELECT ssl, version, cipher + FROM pg_stat_ssl + WHERE pid = pg_backend_pid() + """) + if rows and rows[0]: + ssl_info = rows[0] + ssl_value = ssl_info.get('ssl') + # PostgreSQL returns boolean or 't'/'f' + session.ssl_enabled = ssl_value is True or ssl_value == 't' or ssl_value == 'true' + session.ssl_version = ssl_info.get('version') + session.ssl_cipher = ssl_info.get('cipher') + except Exception: + pass + + # Check if password was used for authentication + session.password_used = bool(getattr(self._backend.config, 'password', None)) + + return session + + def get_wal_info(self) -> WALInfo: + """Get WAL (Write-Ahead Logging) information.""" + wal_info = WALInfo() + + # Get WAL level + try: + rows = self._exec_query("SHOW wal_level") + if rows: + wal_info.wal_level = list(rows[0].values())[0] + except Exception: + pass + + # Get WAL directory size (PostgreSQL 13+) + try: + rows = self._exec_query("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + if rows and rows[0].get('wal_bytes'): + wal_info.wal_size_bytes = int(rows[0]['wal_bytes']) + except Exception: + pass + + # Get WAL files count + try: + rows = self._exec_query(""" + SELECT count(*) as wal_files + FROM pg_ls_waldir() + """) + if rows: + wal_info.wal_files = int(rows[0].get('wal_files', 0)) + except Exception: + # Function may not exist in older versions + pass + + # Get checkpoint statistics + try: + rows = self._exec_query(""" + SELECT checkpoints_timed + checkpoints_req as checkpoint_count + FROM pg_stat_bgwriter + """) + if rows: + wal_info.checkpoint_count = int(rows[0].get('checkpoint_count', 0)) + except Exception: + pass + + # Get last checkpoint time + try: + rows = self._exec_query(""" + SELECT pg_stat_get_xact_checkpoint_timestamp()::text as checkpoint_time + """) + if rows and rows[0].get('checkpoint_time'): + wal_info.checkpoint_time = rows[0]['checkpoint_time'] + except Exception: + pass + + # Get WAL segments (pg_wal files) + try: + rows = self._exec_query("SHOW wal_keep_segments") + if rows: + wal_info.wal_segments = int(list(rows[0].values())[0] or 0) + except Exception: + pass + + return wal_info + + def get_replication_info(self) -> ReplicationInfo: + """Get replication status information.""" + replication_info = ReplicationInfo() + + # Check if primary or standby + try: + rows = self._exec_query("SELECT pg_is_in_recovery()") + if rows: + is_in_recovery = list(rows[0].values())[0] + # pg_is_in_recovery returns False for primary, True for standby + replication_info.is_primary = not (is_in_recovery is True or is_in_recovery == 't' or is_in_recovery == 'true') + replication_info.is_standby = is_in_recovery is True or is_in_recovery == 't' or is_in_recovery == 'true' + except Exception: + pass + + # Get replication slots count + try: + rows = self._exec_query(""" + SELECT count(*) as slot_count + FROM pg_replication_slots + """) + if rows: + replication_info.replication_slots = int(rows[0].get('slot_count', 0)) + except Exception: + pass + + # Get active replications count + try: + rows = self._exec_query(""" + SELECT count(*) as active_count + FROM pg_stat_replication + WHERE state = 'streaming' + """) + if rows: + replication_info.active_replications = int(rows[0].get('active_count', 0)) + replication_info.streaming = replication_info.active_replications > 0 + except Exception: + pass + + # Get replication lag (for standby) + if replication_info.is_standby: + try: + rows = self._exec_query(""" + SELECT pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn())::bigint as lag_bytes + """) + if rows and rows[0].get('lag_bytes') is not None: + replication_info.lag_bytes = int(rows[0]['lag_bytes']) + except Exception: + pass + + return replication_info + + def get_archive_info(self) -> ArchiveInfo: + """Get archive status information.""" + archive_info = ArchiveInfo() + + # Get archive mode + try: + rows = self._exec_query("SHOW archive_mode") + if rows: + archive_info.archive_mode = list(rows[0].values())[0] + except Exception: + pass + + # Get archive command + try: + rows = self._exec_query("SHOW archive_command") + if rows: + archive_info.archive_command = list(rows[0].values())[0] + except Exception: + pass + + # Get archive timeout + try: + rows = self._exec_query("SHOW archive_timeout") + if rows: + timeout_str = list(rows[0].values())[0] + # Parse timeout (could be like '5min' or '300s') + if timeout_str and timeout_str != '0': + # Try to parse as integer (seconds) + try: + archive_info.archive_timeout = int(timeout_str) + except ValueError: + pass + except Exception: + pass + + # Get archive statistics + try: + rows = self._exec_query(""" + SELECT archived_count, failed_count, + last_archived_time::text as last_archived, + last_failed_time::text as last_failed + FROM pg_stat_archiver + """) + if rows: + row = rows[0] + archive_info.archived_count = int(row.get('archived_count', 0) or 0) + archive_info.failed_count = int(row.get('failed_count', 0) or 0) + archive_info.last_archived = row.get('last_archived') + archive_info.last_failed = row.get('last_failed') + except Exception: + pass + + return archive_info + + def get_security_info(self) -> SecurityInfo: + """Get security-related status information.""" + security_info = SecurityInfo() + + # Get SSL status + try: + rows = self._exec_query("SHOW ssl") + if rows: + ssl_value = list(rows[0].values())[0] + security_info.ssl_enabled = ssl_value == 'on' or ssl_value is True + except Exception: + pass + + # Get SSL certificate file + try: + rows = self._exec_query("SHOW ssl_cert_file") + if rows: + security_info.ssl_cert_file = list(rows[0].values())[0] + except Exception: + pass + + # Get SSL key file + try: + rows = self._exec_query("SHOW ssl_key_file") + if rows: + security_info.ssl_key_file = list(rows[0].values())[0] + except Exception: + pass + + # Get SSL CA file + try: + rows = self._exec_query("SHOW ssl_ca_file") + if rows: + security_info.ssl_ca_file = list(rows[0].values())[0] + except Exception: + pass + + # Get password encryption + try: + rows = self._exec_query("SHOW password_encryption") + if rows: + security_info.password_encryption = list(rows[0].values())[0] + except Exception: + pass + + # Check row-level security + try: + rows = self._exec_query("SHOW row_security") + if rows: + rls_value = list(rows[0].values())[0] + security_info.row_security = rls_value == 'on' or rls_value is True + except Exception: + pass + + return security_info + + def list_extensions(self) -> List[ExtensionInfo]: + """List installed extensions.""" + extensions = [] + + try: + rows = self._exec_query(""" + SELECT e.extname as name, + e.extversion as version, + n.nspname as schema, + pg_catalog.obj_description(e.oid, 'pg_extension') as description + FROM pg_extension e + LEFT JOIN pg_namespace n ON n.oid = e.extnamespace + ORDER BY e.extname + """) + + for row in rows: + ext = ExtensionInfo( + name=row['name'], + version=row.get('version'), + schema=row.get('schema'), + description=row.get('description'), + ) + extensions.append(ext) + except Exception: + pass + + return extensions + + +class AsyncPostgreSQLStatusIntrospector( + PostgreSQLStatusIntrospectorMixin, AsyncAbstractStatusIntrospector +): + """Asynchronous PostgreSQL status introspector. + + Usage:: + + backend = AsyncPostgreSQLBackend(connection_config=config) + await backend.connect() + status = await backend.introspector.status.get_overview() + print(status.server_version) + """ + + def __init__(self, backend: Any) -> None: + super().__init__(backend) + + async def _exec_query_async(self, sql: str, params: Optional[tuple] = None) -> List[Dict[str, Any]]: + """Execute a SQL query asynchronously.""" + result = await self._backend.execute(sql, params) + return result.data if result else [] + + async def get_overview(self) -> ServerOverview: + """Get complete PostgreSQL status overview.""" + configuration = await self.list_configuration() + performance = await self.list_performance_metrics() + connections = await self.get_connection_info() + storage = await self.get_storage_info() + databases = await self.list_databases() + users = await self.list_users() + session = await self.get_session_info() + wal = await self.get_wal_info() + replication = await self.get_replication_info() + archive = await self.get_archive_info() + security = await self.get_security_info() + extensions = await self.list_extensions() + + return self._build_server_overview( + configuration=configuration, + performance=performance, + connections=connections, + storage=storage, + databases=databases, + users=users, + session=session, + wal=wal, + replication=replication, + archive=archive, + security=security, + extensions=extensions, + ) + + async def list_configuration( + self, category: Optional[StatusCategory] = None + ) -> List[StatusItem]: + """List PostgreSQL configuration parameters from pg_settings.""" + items = [] + + # Query pg_settings + sql = "SELECT name, setting, unit FROM pg_settings ORDER BY name" + rows = await self._exec_query_async(sql) + + settings_dict = {row['name']: row for row in rows} + + for name, cat, desc, unit, is_readonly in POSTGRES_CONFIG_VARIABLES: + if category and cat != category: + continue + + # Get setting value from pg_settings + row = settings_dict.get(name) + if row: + item = self._create_status_item( + name=name, + value=row.get('setting'), + category=cat, + description=desc, + unit=row.get('unit') if row.get('unit') else unit, + is_readonly=is_readonly, + ) + items.append(item) + + return items + + async def list_performance_metrics( + self, category: Optional[StatusCategory] = None + ) -> List[StatusItem]: + """List PostgreSQL performance metrics from pg_stat_activity.""" + items = [] + + # Query pg_stat_activity for performance-related settings + sql = """ + SELECT name, setting, unit FROM pg_settings + WHERE name IN ('checkpoint_timeout', 'checkpoint_completion_target', 'archive_timeout', 'archive_command') + ORDER BY name + """ + rows = await self._exec_query_async(sql) + + for row in rows: + name = row['name'] + for var_name, var_cat, desc, unit, is_readonly in POSTGRES_CONFIG_VARIABLES: + if name == var_name: + item = self._create_status_item( + name=name, + value=row.get('setting'), + category=var_cat, + description=desc, + unit=row.get('unit') if row.get('unit') else unit, + is_readonly=is_readonly, + ) + items.append(item) + break + + return items + + async def get_connection_info(self) -> ConnectionInfo: + """Get connection information from pg_stat_activity.""" + sql = """ + SELECT count(*) as total_connections + FROM pg_stat_activity + WHERE datname = current_database() + """ + rows = await self._exec_query_async(sql) + total = rows[0].get('total_connections', 1) if rows else 0 + + max_conn = None + try: + sql = "SHOW max_connections" + result = await self._exec_query_async(sql) + if result: + max_conn = int(list(result[0].values())[0]) + except Exception: + pass + + return ConnectionInfo( + active_count=total, + max_connections=max_conn, + ) + + async def get_storage_info(self) -> StorageInfo: + """Get storage information.""" + db_name = getattr(self._backend.config, 'database', 'postgres') + + # Get database size + sql = "SELECT pg_database_size(%s) as size_bytes" + rows = await self._exec_query_async(sql, (db_name,)) + size_bytes = rows[0].get('size_bytes') if rows else None + + return StorageInfo( + total_size_bytes=size_bytes, + ) + + async def list_databases(self) -> List[DatabaseBriefInfo]: + """List databases from pg_database with schema info for current database. + + For the current database, includes schema information with table/view counts + in the extra field. + """ + # Get list of databases + sql = """ + SELECT datname as name, pg_database_size(datname) as size_bytes + FROM pg_database + WHERE datistemplate = false OR datname = current_database() + ORDER BY datname + """ + rows = await self._exec_query_async(sql) + + # Get current database name + current_db = None + try: + result = await self._exec_query_async("SELECT current_database()") + if result: + current_db = result[0].get('current_database') + except Exception: + pass + + # Get schema info for current database + schema_info: Dict[str, Dict[str, int]] = {} + if current_db: + try: + # Query schemata with table/view counts + sql = """ + SELECT + n.nspname as schema_name, + COUNT(CASE WHEN c.relkind = 'r' THEN 1 END) as table_count, + COUNT(CASE WHEN c.relkind = 'v' THEN 1 END) as view_count + FROM pg_namespace n + LEFT JOIN pg_class c ON c.relnamespace = n.oid + WHERE n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast') + GROUP BY n.nspname + ORDER BY n.nspname + """ + schema_rows = await self._exec_query_async(sql) + for row in schema_rows: + schema_name = row.get('schema_name') + if schema_name: + schema_info[schema_name] = { + 'tables': row.get('table_count', 0) or 0, + 'views': row.get('view_count', 0) or 0, + } + except Exception: + pass + + databases = [] + for row in rows: + db_name = row['name'] + db_info = DatabaseBriefInfo( + name=db_name, + size_bytes=row.get('size_bytes'), + ) + + # Add schema info for current database + if db_name == current_db and schema_info: + db_info.extra['schemas'] = schema_info + # Also set total table/view counts + db_info.table_count = sum(s['tables'] for s in schema_info.values()) + db_info.view_count = sum(s['views'] for s in schema_info.values()) + + databases.append(db_info) + + return databases + + async def list_users(self) -> List[UserInfo]: + """List users from pg_roles.""" + sql = """ + SELECT rolname as name, rolsuper as is_superuser, + pg_catalog.pg_get_userbyid(oid) as owner + FROM pg_roles + ORDER BY rolname + """ + rows = await self._exec_query_async(sql) + + return [ + UserInfo( + name=row['name'], + is_superuser=row['is_superuser'], + extra={'owner': row.get('owner')} if row.get('owner') else {}, + ) + for row in rows + ] + + async def get_session_info(self) -> SessionInfo: + """Get current session/connection information.""" + session = SessionInfo() + + # Get current user + try: + rows = await self._exec_query_async("SELECT current_user") + if rows: + session.user = rows[0].get('current_user') + except Exception: + pass + + # Get current database + try: + rows = await self._exec_query_async("SELECT current_database()") + if rows: + session.database = rows[0].get('current_database') + except Exception: + pass + + # Get current schema + try: + rows = await self._exec_query_async("SELECT current_schema()") + if rows: + session.schema = rows[0].get('current_schema') + except Exception: + pass + + # Get client host address + try: + rows = await self._exec_query_async("SELECT inet_client_addr()") + if rows: + host = rows[0].get('inet_client_addr') + if host: + session.host = str(host) + except Exception: + pass + + # Get SSL status + try: + rows = await self._exec_query_async("SELECT ssl_is_used()") + if rows: + ssl_used = rows[0].get('ssl_is_used') + session.ssl_enabled = ssl_used is True or ssl_used == 't' or ssl_used == 'true' + + if session.ssl_enabled: + # Get SSL version + rows = await self._exec_query_async("SELECT ssl_version FROM pg_stat_ssl WHERE pid = pg_backend_pid()") + if rows and rows[0].get('ssl_version'): + session.ssl_version = rows[0].get('ssl_version') + + # Get SSL cipher + rows = await self._exec_query_async("SELECT ssl_cipher FROM pg_stat_ssl WHERE pid = pg_backend_pid()") + if rows and rows[0].get('ssl_cipher'): + session.ssl_cipher = rows[0].get('ssl_cipher') + except Exception: + pass + + # Check if password was used for authentication + session.password_used = bool(getattr(self._backend.config, 'password', None)) + + return session + + async def get_wal_info(self) -> WALInfo: + """Get WAL (Write-Ahead Logging) information.""" + wal_info = WALInfo() + + # Get WAL level + try: + rows = await self._exec_query_async("SHOW wal_level") + if rows: + wal_info.wal_level = list(rows[0].values())[0] + except Exception: + pass + + # Get WAL directory size (PostgreSQL 13+) + try: + rows = await self._exec_query_async("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + if rows and rows[0].get('wal_bytes'): + wal_info.wal_size_bytes = int(rows[0]['wal_bytes']) + except Exception: + pass + + # Get WAL files count + try: + rows = await self._exec_query_async(""" + SELECT count(*) as wal_files + FROM pg_ls_waldir() + """) + if rows: + wal_info.wal_files = int(rows[0].get('wal_files', 0)) + except Exception: + # Function may not exist in older versions + pass + + # Get checkpoint statistics + try: + rows = await self._exec_query_async(""" + SELECT checkpoints_timed + checkpoints_req as checkpoint_count + FROM pg_stat_bgwriter + """) + if rows: + wal_info.checkpoint_count = int(rows[0].get('checkpoint_count', 0)) + except Exception: + pass + + # Get last checkpoint time + try: + rows = await self._exec_query_async(""" + SELECT pg_stat_get_xact_checkpoint_timestamp()::text as checkpoint_time + """) + if rows and rows[0].get('checkpoint_time'): + wal_info.checkpoint_time = rows[0]['checkpoint_time'] + except Exception: + pass + + # Get WAL segments (pg_wal files) + try: + rows = await self._exec_query_async("SHOW wal_keep_segments") + if rows: + wal_info.wal_segments = int(list(rows[0].values())[0] or 0) + except Exception: + pass + + return wal_info + + async def get_replication_info(self) -> ReplicationInfo: + """Get replication status information.""" + replication_info = ReplicationInfo() + + # Check if primary or standby + try: + rows = await self._exec_query_async("SELECT pg_is_in_recovery()") + if rows: + is_in_recovery = list(rows[0].values())[0] + # pg_is_in_recovery returns False for primary, True for standby + replication_info.is_primary = not (is_in_recovery is True or is_in_recovery == 't' or is_in_recovery == 'true') + replication_info.is_standby = is_in_recovery is True or is_in_recovery == 't' or is_in_recovery == 'true' + except Exception: + pass + + # Get replication slots count + try: + rows = await self._exec_query_async(""" + SELECT count(*) as slot_count + FROM pg_replication_slots + """) + if rows: + replication_info.replication_slots = int(rows[0].get('slot_count', 0)) + except Exception: + pass + + # Get active replications count + try: + rows = await self._exec_query_async(""" + SELECT count(*) as active_count + FROM pg_stat_replication + WHERE state = 'streaming' + """) + if rows: + replication_info.active_replications = int(rows[0].get('active_count', 0)) + replication_info.streaming = replication_info.active_replications > 0 + except Exception: + pass + + # Get replication lag (for standby) + if replication_info.is_standby: + try: + rows = await self._exec_query_async(""" + SELECT pg_wal_lsn_diff(pg_last_wal_receive_lsn(), pg_last_wal_replay_lsn())::bigint as lag_bytes + """) + if rows and rows[0].get('lag_bytes') is not None: + replication_info.lag_bytes = int(rows[0]['lag_bytes']) + except Exception: + pass + + return replication_info + + async def get_archive_info(self) -> ArchiveInfo: + """Get archive status information.""" + archive_info = ArchiveInfo() + + # Get archive mode + try: + rows = await self._exec_query_async("SHOW archive_mode") + if rows: + archive_info.archive_mode = list(rows[0].values())[0] + except Exception: + pass + + # Get archive command + try: + rows = await self._exec_query_async("SHOW archive_command") + if rows: + archive_info.archive_command = list(rows[0].values())[0] + except Exception: + pass + + # Get archive timeout + try: + rows = await self._exec_query_async("SHOW archive_timeout") + if rows: + timeout_str = list(rows[0].values())[0] + # Parse timeout (could be like '5min' or '300s') + if timeout_str and timeout_str != '0': + # Try to parse as integer (seconds) + try: + archive_info.archive_timeout = int(timeout_str) + except ValueError: + pass + except Exception: + pass + + # Get archive statistics + try: + rows = await self._exec_query_async(""" + SELECT archived_count, failed_count, + last_archived_time::text as last_archived, + last_failed_time::text as last_failed + FROM pg_stat_archiver + """) + if rows: + row = rows[0] + archive_info.archived_count = int(row.get('archived_count', 0) or 0) + archive_info.failed_count = int(row.get('failed_count', 0) or 0) + archive_info.last_archived = row.get('last_archived') + archive_info.last_failed = row.get('last_failed') + except Exception: + pass + + return archive_info + + async def get_security_info(self) -> SecurityInfo: + """Get security-related status information.""" + security_info = SecurityInfo() + + # Get SSL status + try: + rows = await self._exec_query_async("SHOW ssl") + if rows: + ssl_value = list(rows[0].values())[0] + security_info.ssl_enabled = ssl_value == 'on' or ssl_value is True + except Exception: + pass + + # Get SSL certificate file + try: + rows = await self._exec_query_async("SHOW ssl_cert_file") + if rows: + security_info.ssl_cert_file = list(rows[0].values())[0] + except Exception: + pass + + # Get SSL key file + try: + rows = await self._exec_query_async("SHOW ssl_key_file") + if rows: + security_info.ssl_key_file = list(rows[0].values())[0] + except Exception: + pass + + # Get SSL CA file + try: + rows = await self._exec_query_async("SHOW ssl_ca_file") + if rows: + security_info.ssl_ca_file = list(rows[0].values())[0] + except Exception: + pass + + # Get password encryption + try: + rows = await self._exec_query_async("SHOW password_encryption") + if rows: + security_info.password_encryption = list(rows[0].values())[0] + except Exception: + pass + + # Check row-level security + try: + rows = await self._exec_query_async("SHOW row_security") + if rows: + rls_value = list(rows[0].values())[0] + security_info.row_security = rls_value == 'on' or rls_value is True + except Exception: + pass + + return security_info + + async def list_extensions(self) -> List[ExtensionInfo]: + """List installed extensions.""" + extensions = [] + + try: + rows = await self._exec_query_async(""" + SELECT e.extname as name, + e.extversion as version, + n.nspname as schema, + pg_catalog.obj_description(e.oid, 'pg_extension') as description + FROM pg_extension e + LEFT JOIN pg_namespace n ON n.oid = e.extnamespace + ORDER BY e.extname + """) + + for row in rows: + ext = ExtensionInfo( + name=row['name'], + version=row.get('version'), + schema=row.get('schema'), + description=row.get('description'), + ) + extensions.append(ext) + except Exception: + pass + + return extensions From 46f04e55b63d855fa045fced733331e879ed85c2 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 8 Apr 2026 12:35:42 +0800 Subject: [PATCH 2/4] fix(introspection): resolve PostgreSQL version compatibility issues Fixed WAL info queries to work across PostgreSQL 9.6 to 18: - PG 9.6 uses pg_current_xlog_location() instead of pg_current_wal_lsn() - PG 9.6 uses pg_xlog_location_diff() instead of pg_wal_lsn_diff() - pg_ls_waldir() only exists in PostgreSQL 10+ - PG 17+ uses pg_stat_checkpointer for checkpoint statistics - Use pg_control_checkpoint() for checkpoint LSN info - PG 13+ uses wal_keep_size, older versions use wal_keep_segments Added _get_version_tuple() method for version comparison. Co-Authored-By: Claude Opus 4.6 --- .../introspection/status_introspector.py | 258 +++++++++++++----- 1 file changed, 192 insertions(+), 66 deletions(-) diff --git a/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py b/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py index 7d4d2845..156f766b 100644 --- a/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py +++ b/src/rhosocial/activerecord/backend/impl/postgres/introspection/status_introspector.py @@ -43,8 +43,8 @@ ("maintenance_work_mem", StatusCategory.CONFIGURATION, "Maintenance work memory", "bytes", True), ("dynamic_shared_memory", StatusCategory.CONFIGURATION, "Dynamic shared memory", None, False), ("wal_level", StatusCategory.CONFIGURATION, "WAL level", None, False), - ("wal_keep_segments", StatusCategory.CONFIGURATION, "WAL segments", None, False), - ("synchronous_commit", StatusCategory.CONFIGURATION, "Synchronous commit", None, False), + ("wal_keep_size", StatusCategory.CONFIGURATION, "WAL keep size", "bytes", False), # Changed from wal_keep_segments + ("synchronous_commit", StatusCategory.PERFORMANCE, "Synchronous commit", None, False), ("checkpoint_timeout", StatusCategory.PERFORMANCE, "Checkpoint timeout", "seconds", False), ("checkpoint_completion_target", StatusCategory.PERFORMANCE, "Checkpoint completion target", "seconds", False), ("archive_timeout", StatusCategory.PERFORMANCE, "Archive timeout", "seconds", False), @@ -67,6 +67,29 @@ ] +# Performance metrics from pg_stat_bgwriter and other views +# Format: (variable_name, category, description, unit) +POSTGRES_STATUS_VARIABLES = [ + # Checkpoint statistics (available in PostgreSQL 10-16, removed in 17+) + ("checkpoints_timed", StatusCategory.PERFORMANCE, "Scheduled checkpoints", "checkpoints"), + ("checkpoints_req", StatusCategory.PERFORMANCE, "Requested checkpoints", "checkpoints"), + + # Buffer pool statistics + ("buffers_alloc", StatusCategory.PERFORMANCE, "Buffer allocations", None), + ("buffers_backend_fsync", StatusCategory.PERFORMANCE, "Buffer backend fsync", None), + ("buffers_checkpoint", StatusCategory.PERFORMANCE, "Buffers written by checkpoint", None), + + # Connection statistics (from pg_stat_activity) + ("numbackends", StatusCategory.CONNECTION, "Number of backend connections", None), + ("sessions", StatusCategory.CONNECTION, "Number of sessions", None), + ("active", StatusCategory.CONNECTION, "Number of active sessions", None), + ("idle", StatusCategory.CONNECTION, "Number of idle sessions", None), + ("waiting", StatusCategory.CONNECTION, "Number of waiting sessions", None), + ("total", StatusCategory.CONNECTION, "Total connections", None), + ("max_conn", StatusCategory.CONNECTION, "Maximum connections", None), +] + + class PostgreSQLStatusIntrospectorMixin: """Mixin providing shared PostgreSQL status introspection logic. @@ -131,6 +154,37 @@ def _get_version_string(self) -> str: return "Unknown" + def _get_version_tuple(self) -> tuple: + """Get PostgreSQL version as tuple for comparison. + + Returns tuple like (18, 3, 0) or (9, 6, 24). + """ + # Try to get version from backend's _server_version_cache attribute first + version = getattr(self._backend, '_server_version_cache', None) + if version: + return tuple(version) + + # Fall back to _version attribute + version = getattr(self._backend, '_version', None) + if version: + return tuple(version) + + # Try dialect version + if hasattr(self._backend, '_dialect') and hasattr(self._backend._dialect, 'version'): + version = self._backend._dialect.version + if version: + return tuple(version) + + # Last resort: query server directly + try: + version = self._backend.get_server_version() + if version: + return tuple(version) + except Exception: + pass + + return (0, 0, 0) # Unknown version + def _create_status_item( self, name: str, @@ -485,6 +539,7 @@ def get_session_info(self) -> SessionInfo: def get_wal_info(self) -> WALInfo: """Get WAL (Write-Ahead Logging) information.""" wal_info = WALInfo() + version = self._get_version_tuple() # Get WAL level try: @@ -494,54 +549,89 @@ def get_wal_info(self) -> WALInfo: except Exception: pass - # Get WAL directory size (PostgreSQL 13+) + # Get WAL position/size + # PostgreSQL 10+: pg_current_wal_lsn() + # PostgreSQL 9.6 and below: pg_current_xlog_location() try: - rows = self._exec_query("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + if version >= (10, 0, 0): + rows = self._exec_query("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + else: + # PG 9.6 and below use xlog terminology + rows = self._exec_query("SELECT pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint as wal_bytes") if rows and rows[0].get('wal_bytes'): wal_info.wal_size_bytes = int(rows[0]['wal_bytes']) except Exception: pass # Get WAL files count - try: - rows = self._exec_query(""" - SELECT count(*) as wal_files - FROM pg_ls_waldir() - """) - if rows: - wal_info.wal_files = int(rows[0].get('wal_files', 0)) - except Exception: - # Function may not exist in older versions - pass + # pg_ls_waldir() only exists in PostgreSQL 10+ + if version >= (10, 0, 0): + try: + rows = self._exec_query(""" + SELECT count(*) as wal_files + FROM pg_ls_waldir() + """) + if rows: + wal_info.wal_files = int(rows[0].get('wal_files', 0)) + except Exception: + pass # Get checkpoint statistics + # PostgreSQL 10-16: pg_stat_bgwriter has checkpoints_timed and checkpoints_req + # PostgreSQL 17+: these columns moved to pg_stat_checkpointer try: - rows = self._exec_query(""" - SELECT checkpoints_timed + checkpoints_req as checkpoint_count - FROM pg_stat_bgwriter - """) + if version >= (17, 0, 0): + # PostgreSQL 17+ uses pg_stat_checkpointer + rows = self._exec_query(""" + SELECT + COALESCE(num_timed, 0) + + COALESCE(num_requested, 0) as checkpoint_count + FROM pg_stat_checkpointer + """) + else: + # PostgreSQL 10-16 uses pg_stat_bgwriter + rows = self._exec_query(""" + SELECT + COALESCE(checkpoints_timed, 0) + + COALESCE(checkpoints_req, 0) as checkpoint_count + FROM pg_stat_bgwriter + """) if rows: wal_info.checkpoint_count = int(rows[0].get('checkpoint_count', 0)) except Exception: pass - # Get last checkpoint time - try: - rows = self._exec_query(""" - SELECT pg_stat_get_xact_checkpoint_timestamp()::text as checkpoint_time - """) - if rows and rows[0].get('checkpoint_time'): - wal_info.checkpoint_time = rows[0]['checkpoint_time'] - except Exception: - pass + # Get last checkpoint info + # pg_control_checkpoint() exists in PostgreSQL 9.6+ + # Returns a record with checkpoint LSN as the first column + if version >= (10, 0, 0): + try: + rows = self._exec_query(""" + SELECT (pg_control_checkpoint()).checkpoint_lsn as checkpoint_info + """) + if rows and rows[0].get('checkpoint_info'): + wal_info.checkpoint_time = rows[0]['checkpoint_info'] + except Exception: + # Some PostgreSQL versions may not support this syntax + pass - # Get WAL segments (pg_wal files) - try: - rows = self._exec_query("SHOW wal_keep_segments") - if rows: - wal_info.wal_segments = int(list(rows[0].values())[0] or 0) - except Exception: - pass + # Get WAL keep size/segments + # PostgreSQL 13+: wal_keep_size + # PostgreSQL 12 and below: wal_keep_segments + if version >= (13, 0, 0): + try: + rows = self._exec_query("SHOW wal_keep_size") + if rows: + wal_info.wal_segments = list(rows[0].values())[0] + except Exception: + pass + else: + try: + rows = self._exec_query("SHOW wal_keep_segments") + if rows: + wal_info.wal_segments = list(rows[0].values())[0] + except Exception: + pass return wal_info @@ -1041,6 +1131,7 @@ async def get_session_info(self) -> SessionInfo: async def get_wal_info(self) -> WALInfo: """Get WAL (Write-Ahead Logging) information.""" wal_info = WALInfo() + version = self._get_version_tuple() # Get WAL level try: @@ -1050,54 +1141,89 @@ async def get_wal_info(self) -> WALInfo: except Exception: pass - # Get WAL directory size (PostgreSQL 13+) + # Get WAL position/size + # PostgreSQL 10+: pg_current_wal_lsn() + # PostgreSQL 9.6 and below: pg_current_xlog_location() try: - rows = await self._exec_query_async("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + if version >= (10, 0, 0): + rows = await self._exec_query_async("SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0')::bigint as wal_bytes") + else: + # PG 9.6 and below use xlog terminology + rows = await self._exec_query_async("SELECT pg_xlog_location_diff(pg_current_xlog_location(), '0/0')::bigint as wal_bytes") if rows and rows[0].get('wal_bytes'): wal_info.wal_size_bytes = int(rows[0]['wal_bytes']) except Exception: pass # Get WAL files count - try: - rows = await self._exec_query_async(""" - SELECT count(*) as wal_files - FROM pg_ls_waldir() - """) - if rows: - wal_info.wal_files = int(rows[0].get('wal_files', 0)) - except Exception: - # Function may not exist in older versions - pass + # pg_ls_waldir() only exists in PostgreSQL 10+ + if version >= (10, 0, 0): + try: + rows = await self._exec_query_async(""" + SELECT count(*) as wal_files + FROM pg_ls_waldir() + """) + if rows: + wal_info.wal_files = int(rows[0].get('wal_files', 0)) + except Exception: + pass # Get checkpoint statistics + # PostgreSQL 10-16: pg_stat_bgwriter has checkpoints_timed and checkpoints_req + # PostgreSQL 17+: these columns moved to pg_stat_checkpointer try: - rows = await self._exec_query_async(""" - SELECT checkpoints_timed + checkpoints_req as checkpoint_count - FROM pg_stat_bgwriter - """) + if version >= (17, 0, 0): + # PostgreSQL 17+ uses pg_stat_checkpointer + rows = await self._exec_query_async(""" + SELECT + COALESCE(num_timed, 0) + + COALESCE(num_requested, 0) as checkpoint_count + FROM pg_stat_checkpointer + """) + else: + # PostgreSQL 10-16 uses pg_stat_bgwriter + rows = await self._exec_query_async(""" + SELECT + COALESCE(checkpoints_timed, 0) + + COALESCE(checkpoints_req, 0) as checkpoint_count + FROM pg_stat_bgwriter + """) if rows: wal_info.checkpoint_count = int(rows[0].get('checkpoint_count', 0)) except Exception: pass - # Get last checkpoint time - try: - rows = await self._exec_query_async(""" - SELECT pg_stat_get_xact_checkpoint_timestamp()::text as checkpoint_time - """) - if rows and rows[0].get('checkpoint_time'): - wal_info.checkpoint_time = rows[0]['checkpoint_time'] - except Exception: - pass + # Get last checkpoint info + # pg_control_checkpoint() exists in PostgreSQL 9.6+ + # Returns a record with checkpoint LSN as the first column + if version >= (10, 0, 0): + try: + rows = await self._exec_query_async(""" + SELECT (pg_control_checkpoint()).checkpoint_lsn as checkpoint_info + """) + if rows and rows[0].get('checkpoint_info'): + wal_info.checkpoint_time = rows[0]['checkpoint_info'] + except Exception: + # Some PostgreSQL versions may not support this syntax + pass - # Get WAL segments (pg_wal files) - try: - rows = await self._exec_query_async("SHOW wal_keep_segments") - if rows: - wal_info.wal_segments = int(list(rows[0].values())[0] or 0) - except Exception: - pass + # Get WAL keep size/segments + # PostgreSQL 13+: wal_keep_size + # PostgreSQL 12 and below: wal_keep_segments + if version >= (13, 0, 0): + try: + rows = await self._exec_query_async("SHOW wal_keep_size") + if rows: + wal_info.wal_segments = list(rows[0].values())[0] + except Exception: + pass + else: + try: + rows = await self._exec_query_async("SHOW wal_keep_segments") + if rows: + wal_info.wal_segments = list(rows[0].values())[0] + except Exception: + pass return wal_info From f8f2e20cab7b79e362cbcadc8a18637930fdd333 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 8 Apr 2026 13:15:33 +0800 Subject: [PATCH 3/4] test(backend): add status introspector test coverage for PostgreSQL Add comprehensive tests for PostgreSQL status introspector including: - get_overview, list_configuration, list_performance_metrics - get_connection_info, get_storage_info, list_databases, list_users - PostgreSQL-specific: get_wal_info, get_replication_info, get_archive_info, get_security_info, list_extensions Both synchronous and asynchronous API tests are included. Co-Authored-By: Claude Opus 4.6 --- .../introspection/test_status_introspector.py | 624 ++++++++++++++++++ 1 file changed, 624 insertions(+) create mode 100644 tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_status_introspector.py diff --git a/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_status_introspector.py b/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_status_introspector.py new file mode 100644 index 00000000..b592f5a4 --- /dev/null +++ b/tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_status_introspector.py @@ -0,0 +1,624 @@ +# tests/rhosocial/activerecord_postgres_test/feature/backend/introspection/test_status_introspector.py +""" +Tests for PostgreSQL status introspector. + +This module tests the SyncPostgreSQLStatusIntrospector functionality +for retrieving server status information via pg_settings, pg_stat_activity, +and other system views. +""" + +import pytest + +from rhosocial.activerecord.backend.introspection.status import ( + StatusItem, + StatusCategory, + ServerOverview, + DatabaseBriefInfo, + UserInfo, + ConnectionInfo, + StorageInfo, + WALInfo, + ReplicationInfo, + ArchiveInfo, + SecurityInfo, + ExtensionInfo, +) + + +class TestSyncPostgreSQLStatusIntrospector: + """Tests for synchronous PostgreSQL status introspector.""" + + def test_get_overview(self, postgres_backend_single): + """Test get_overview returns valid ServerOverview.""" + status = postgres_backend_single.introspector.status + + overview = status.get_overview() + + assert isinstance(overview, ServerOverview) + assert overview.server_vendor == "PostgreSQL" + assert overview.server_version is not None + assert isinstance(overview.configuration, list) + assert isinstance(overview.performance, list) + assert isinstance(overview.storage, StorageInfo) + assert isinstance(overview.databases, list) + # PostgreSQL has users + assert isinstance(overview.users, list) + + def test_get_overview_version_matches_dialect(self, postgres_backend_single): + """Test that overview version matches dialect version.""" + status = postgres_backend_single.introspector.status + overview = status.get_overview() + + expected_version = ".".join(map(str, postgres_backend_single.dialect.version)) + assert overview.server_version == expected_version + + def test_get_overview_contains_postgresql_version_info(self, postgres_backend_single): + """Test that overview contains PostgreSQL version info in extra.""" + status = postgres_backend_single.introspector.status + overview = status.get_overview() + + # PostgreSQL should have version info + assert overview.server_version is not None + + def test_list_configuration(self, postgres_backend_single): + """Test list_configuration returns configuration items.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration() + + assert isinstance(items, list) + assert len(items) > 0 + + # Check that all items are StatusItem instances + for item in items: + assert isinstance(item, StatusItem) + assert item.name is not None + assert item.value is not None + + def test_list_configuration_with_category_filter(self, postgres_backend_single): + """Test list_configuration with category filter.""" + status = postgres_backend_single.introspector.status + + config_items = status.list_configuration(category=StatusCategory.CONFIGURATION) + + for item in config_items: + assert item.category == StatusCategory.CONFIGURATION + + def test_list_configuration_contains_expected_items(self, postgres_backend_single): + """Test that configuration contains expected PostgreSQL settings.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration() + item_names = [item.name for item in items] + + # Check for some common PostgreSQL settings + assert "port" in item_names + assert "max_connections" in item_names + + def test_list_configuration_values_are_parsed(self, postgres_backend_single): + """Test that configuration values are returned.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration() + + # port should exist and have a value + port_item = next((i for i in items if i.name == "port"), None) + if port_item: + # PostgreSQL's _parse_setting_value preserves string values + assert port_item.value is not None + + def test_list_performance_metrics(self, postgres_backend_single): + """Test list_performance_metrics returns performance items.""" + status = postgres_backend_single.introspector.status + + items = status.list_performance_metrics() + + assert isinstance(items, list) + + # All items should be PERFORMANCE category + for item in items: + assert item.category == StatusCategory.PERFORMANCE + + def test_get_connection_info(self, postgres_backend_single): + """Test get_connection_info returns ConnectionInfo.""" + status = postgres_backend_single.introspector.status + + conn_info = status.get_connection_info() + + assert isinstance(conn_info, ConnectionInfo) + # PostgreSQL has connection info + assert conn_info.active_count is not None or conn_info.max_connections is not None + + def test_get_storage_info(self, postgres_backend_single): + """Test get_storage_info returns StorageInfo.""" + status = postgres_backend_single.introspector.status + + storage = status.get_storage_info() + + assert isinstance(storage, StorageInfo) + # PostgreSQL should have storage info + assert storage.extra is not None + + def test_list_databases(self, postgres_backend_single): + """Test list_databases returns database list.""" + status = postgres_backend_single.introspector.status + + databases = status.list_databases() + + assert isinstance(databases, list) + assert len(databases) >= 1 + + # All items should be DatabaseBriefInfo instances + for db in databases: + assert isinstance(db, DatabaseBriefInfo) + assert db.name is not None + + def test_list_databases_with_tables(self, backend_with_tables): + """Test list_databases includes table count.""" + status = backend_with_tables.introspector.status + + databases = status.list_databases() + + # At least one database should have tables + assert len(databases) >= 1 + + def test_list_users(self, postgres_backend_single): + """Test list_users returns user list.""" + status = postgres_backend_single.introspector.status + + users = status.list_users() + + assert isinstance(users, list) + + # PostgreSQL typically has at least one user + for user in users: + assert isinstance(user, UserInfo) + assert user.name is not None + + def test_get_wal_info(self, postgres_backend_single): + """Test get_wal_info returns WALInfo.""" + status = postgres_backend_single.introspector.status + + wal_info = status.get_wal_info() + + assert isinstance(wal_info, WALInfo) + # PostgreSQL should have WAL info + assert wal_info.extra is not None + + def test_get_replication_info(self, postgres_backend_single): + """Test get_replication_info returns ReplicationInfo.""" + status = postgres_backend_single.introspector.status + + repl_info = status.get_replication_info() + + assert isinstance(repl_info, ReplicationInfo) + # Replication may or may not be configured + + def test_get_archive_info(self, postgres_backend_single): + """Test get_archive_info returns ArchiveInfo.""" + status = postgres_backend_single.introspector.status + + archive_info = status.get_archive_info() + + assert isinstance(archive_info, ArchiveInfo) + # Archive info should be available + + def test_get_security_info(self, postgres_backend_single): + """Test get_security_info returns SecurityInfo.""" + status = postgres_backend_single.introspector.status + + security_info = status.get_security_info() + + assert isinstance(security_info, SecurityInfo) + # Security info should be available + + def test_list_extensions(self, postgres_backend_single): + """Test list_extensions returns extension list.""" + status = postgres_backend_single.introspector.status + + extensions = status.list_extensions() + + assert isinstance(extensions, list) + + # All items should be ExtensionInfo instances + for ext in extensions: + assert isinstance(ext, ExtensionInfo) + assert ext.name is not None + + def test_status_item_has_description(self, postgres_backend_single): + """Test that status items have descriptions.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration() + + # Check that items have descriptions + for item in items: + assert item.description is not None + + def test_status_item_readonly_flag(self, postgres_backend_single): + """Test that readonly items are marked correctly.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration() + + # port should be readonly (requires restart) + port_item = next((i for i in items if i.name == "port"), None) + if port_item: + assert port_item.is_readonly is True + + +class TestPostgreSQLStatusIntrospectorMixin: + """Tests for PostgreSQLStatusIntrospectorMixin helper methods.""" + + def test_parse_setting_value_int(self, postgres_backend_single): + """Test _parse_setting_value handles integers.""" + status = postgres_backend_single.introspector.status + + # _parse_setting_value returns int if input is int, otherwise str + result = status._parse_setting_value(42) + assert result == 42 + assert isinstance(result, int) + + # String "42" is not converted to int + result_str = status._parse_setting_value("42") + assert result_str == "42" + assert isinstance(result_str, str) + + def test_parse_setting_value_str(self, postgres_backend_single): + """Test _parse_setting_value preserves non-integer strings.""" + status = postgres_backend_single.introspector.status + + result = status._parse_setting_value("utf8") + assert result == "utf8" + assert isinstance(result, str) + + def test_parse_setting_value_on_off(self, postgres_backend_single): + """Test _parse_setting_value handles on/off values.""" + status = postgres_backend_single.introspector.status + + result_on = status._parse_setting_value("on") + assert result_on is True + + result_off = status._parse_setting_value("off") + assert result_off is False + + def test_create_status_item(self, postgres_backend_single): + """Test _create_status_item creates proper StatusItem.""" + status = postgres_backend_single.introspector.status + + item = status._create_status_item( + name="test_param", + value="42", + category=StatusCategory.CONFIGURATION, + description="Test parameter", + unit="ms", + is_readonly=False, + ) + + assert isinstance(item, StatusItem) + assert item.name == "test_param" + # PostgreSQL's _parse_setting_value preserves string "42" + assert item.value == "42" + assert item.category == StatusCategory.CONFIGURATION + assert item.description == "Test parameter" + assert item.unit == "ms" + assert item.is_readonly is False + + def test_get_vendor_name(self, postgres_backend_single): + """Test _get_vendor_name returns PostgreSQL.""" + status = postgres_backend_single.introspector.status + + vendor = status._get_vendor_name() + assert vendor == "PostgreSQL" + + +class TestStatusIntrospectorCategories: + """Tests for different status categories.""" + + def test_configuration_category_items(self, postgres_backend_single): + """Test items in CONFIGURATION category.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration(category=StatusCategory.CONFIGURATION) + + for item in items: + assert item.category == StatusCategory.CONFIGURATION + + def test_performance_category_items(self, postgres_backend_single): + """Test items in PERFORMANCE category.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration(category=StatusCategory.PERFORMANCE) + + for item in items: + assert item.category == StatusCategory.PERFORMANCE + + def test_storage_category_items(self, postgres_backend_single): + """Test items in STORAGE category.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration(category=StatusCategory.STORAGE) + + for item in items: + assert item.category == StatusCategory.STORAGE + + def test_security_category_items(self, postgres_backend_single): + """Test items in SECURITY category.""" + status = postgres_backend_single.introspector.status + + items = status.list_configuration(category=StatusCategory.SECURITY) + + for item in items: + assert item.category == StatusCategory.SECURITY + + +class TestAsyncPostgreSQLStatusIntrospector: + """Tests for asynchronous PostgreSQL status introspector.""" + + @pytest.mark.asyncio + async def test_get_overview(self, async_postgres_backend_single): + """Test async get_overview returns valid ServerOverview.""" + status = async_postgres_backend_single.introspector.status + + overview = await status.get_overview() + + assert isinstance(overview, ServerOverview) + assert overview.server_vendor == "PostgreSQL" + assert overview.server_version is not None + assert isinstance(overview.configuration, list) + assert isinstance(overview.performance, list) + assert isinstance(overview.storage, StorageInfo) + assert isinstance(overview.databases, list) + + @pytest.mark.asyncio + async def test_get_overview_version_matches_dialect(self, async_postgres_backend_single): + """Test that async overview version matches dialect version.""" + status = async_postgres_backend_single.introspector.status + overview = await status.get_overview() + + expected_version = ".".join(map(str, async_postgres_backend_single.dialect.version)) + assert overview.server_version == expected_version + + @pytest.mark.asyncio + async def test_list_configuration(self, async_postgres_backend_single): + """Test async list_configuration returns configuration items.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration() + + assert isinstance(items, list) + assert len(items) > 0 + + # Check that all items are StatusItem instances + for item in items: + assert isinstance(item, StatusItem) + assert item.name is not None + assert item.value is not None + + @pytest.mark.asyncio + async def test_list_configuration_with_category_filter(self, async_postgres_backend_single): + """Test async list_configuration with category filter.""" + status = async_postgres_backend_single.introspector.status + + config_items = await status.list_configuration(category=StatusCategory.CONFIGURATION) + + for item in config_items: + assert item.category == StatusCategory.CONFIGURATION + + @pytest.mark.asyncio + async def test_list_configuration_contains_expected_items(self, async_postgres_backend_single): + """Test that async configuration contains expected PostgreSQL settings.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration() + item_names = [item.name for item in items] + + # Check for some common PostgreSQL settings + assert "port" in item_names + assert "max_connections" in item_names + + @pytest.mark.asyncio + async def test_list_performance_metrics(self, async_postgres_backend_single): + """Test async list_performance_metrics returns performance items.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_performance_metrics() + + assert isinstance(items, list) + + # All items should be PERFORMANCE category + for item in items: + assert item.category == StatusCategory.PERFORMANCE + + @pytest.mark.asyncio + async def test_get_connection_info(self, async_postgres_backend_single): + """Test async get_connection_info returns ConnectionInfo.""" + status = async_postgres_backend_single.introspector.status + + conn_info = await status.get_connection_info() + + assert isinstance(conn_info, ConnectionInfo) + assert conn_info.active_count is not None or conn_info.max_connections is not None + + @pytest.mark.asyncio + async def test_get_storage_info(self, async_postgres_backend_single): + """Test async get_storage_info returns StorageInfo.""" + status = async_postgres_backend_single.introspector.status + + storage = await status.get_storage_info() + + assert isinstance(storage, StorageInfo) + + @pytest.mark.asyncio + async def test_list_databases(self, async_postgres_backend_single): + """Test async list_databases returns database list.""" + status = async_postgres_backend_single.introspector.status + + databases = await status.list_databases() + + assert isinstance(databases, list) + assert len(databases) >= 1 + + for db in databases: + assert isinstance(db, DatabaseBriefInfo) + + @pytest.mark.asyncio + async def test_list_users(self, async_postgres_backend_single): + """Test async list_users returns user list.""" + status = async_postgres_backend_single.introspector.status + + users = await status.list_users() + + assert isinstance(users, list) + + for user in users: + assert isinstance(user, UserInfo) + + @pytest.mark.asyncio + async def test_get_wal_info(self, async_postgres_backend_single): + """Test async get_wal_info returns WALInfo.""" + status = async_postgres_backend_single.introspector.status + + wal_info = await status.get_wal_info() + + assert isinstance(wal_info, WALInfo) + + @pytest.mark.asyncio + async def test_get_replication_info(self, async_postgres_backend_single): + """Test async get_replication_info returns ReplicationInfo.""" + status = async_postgres_backend_single.introspector.status + + repl_info = await status.get_replication_info() + + assert isinstance(repl_info, ReplicationInfo) + + @pytest.mark.asyncio + async def test_get_archive_info(self, async_postgres_backend_single): + """Test async get_archive_info returns ArchiveInfo.""" + status = async_postgres_backend_single.introspector.status + + archive_info = await status.get_archive_info() + + assert isinstance(archive_info, ArchiveInfo) + + @pytest.mark.asyncio + async def test_get_security_info(self, async_postgres_backend_single): + """Test async get_security_info returns SecurityInfo.""" + status = async_postgres_backend_single.introspector.status + + security_info = await status.get_security_info() + + assert isinstance(security_info, SecurityInfo) + + @pytest.mark.asyncio + async def test_list_extensions(self, async_postgres_backend_single): + """Test async list_extensions returns extension list.""" + status = async_postgres_backend_single.introspector.status + + extensions = await status.list_extensions() + + assert isinstance(extensions, list) + + for ext in extensions: + assert isinstance(ext, ExtensionInfo) + + @pytest.mark.asyncio + async def test_status_item_has_description(self, async_postgres_backend_single): + """Test that async status items have descriptions.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration() + + for item in items: + assert item.description is not None + + @pytest.mark.asyncio + async def test_status_item_readonly_flag(self, async_postgres_backend_single): + """Test that async readonly items are marked correctly.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration() + + port_item = next((i for i in items if i.name == "port"), None) + if port_item: + assert port_item.is_readonly is True + + +class TestAsyncPostgreSQLStatusIntrospectorMixin: + """Tests for async PostgreSQLStatusIntrospectorMixin helper methods.""" + + @pytest.mark.asyncio + async def test_parse_setting_value_int(self, async_postgres_backend_single): + """Test _parse_setting_value handles integers.""" + status = async_postgres_backend_single.introspector.status + + # _parse_setting_value returns int if input is int, otherwise str + result = status._parse_setting_value(42) + assert result == 42 + assert isinstance(result, int) + + # String "42" is not converted to int + result_str = status._parse_setting_value("42") + assert result_str == "42" + assert isinstance(result_str, str) + + @pytest.mark.asyncio + async def test_parse_setting_value_str(self, async_postgres_backend_single): + """Test _parse_setting_value preserves non-integer strings.""" + status = async_postgres_backend_single.introspector.status + + result = status._parse_setting_value("utf8") + assert result == "utf8" + assert isinstance(result, str) + + @pytest.mark.asyncio + async def test_get_vendor_name(self, async_postgres_backend_single): + """Test _get_vendor_name returns PostgreSQL.""" + status = async_postgres_backend_single.introspector.status + + vendor = status._get_vendor_name() + assert vendor == "PostgreSQL" + + +class TestAsyncStatusIntrospectorCategories: + """Tests for different async status categories.""" + + @pytest.mark.asyncio + async def test_configuration_category_items(self, async_postgres_backend_single): + """Test async items in CONFIGURATION category.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration(category=StatusCategory.CONFIGURATION) + + for item in items: + assert item.category == StatusCategory.CONFIGURATION + + @pytest.mark.asyncio + async def test_performance_category_items(self, async_postgres_backend_single): + """Test async items in PERFORMANCE category.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration(category=StatusCategory.PERFORMANCE) + + for item in items: + assert item.category == StatusCategory.PERFORMANCE + + @pytest.mark.asyncio + async def test_storage_category_items(self, async_postgres_backend_single): + """Test async items in STORAGE category.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration(category=StatusCategory.STORAGE) + + for item in items: + assert item.category == StatusCategory.STORAGE + + @pytest.mark.asyncio + async def test_security_category_items(self, async_postgres_backend_single): + """Test async items in SECURITY category.""" + status = async_postgres_backend_single.introspector.status + + items = await status.list_configuration(category=StatusCategory.SECURITY) + + for item in items: + assert item.category == StatusCategory.SECURITY From 85ba99dfc4436a0b571dd121ba5f7dde9126a533 Mon Sep 17 00:00:00 2001 From: zhao vistart Date: Wed, 8 Apr 2026 13:47:38 +0800 Subject: [PATCH 4/4] chore: add changelog fragment for PR #21 Co-Authored-By: Claude Opus 4.6 --- changelog.d/21.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/21.added.md diff --git a/changelog.d/21.added.md b/changelog.d/21.added.md new file mode 100644 index 00000000..9cf5ced4 --- /dev/null +++ b/changelog.d/21.added.md @@ -0,0 +1 @@ +Added PostgreSQL server status overview support with comprehensive introspection capabilities including WAL, replication, archive, security, and extensions info.