analyze_schema_relationships

Instructions

Implementation Reference

• src/postgres_mcp_pro_plus/server.py:686-714 (registration)

  MCP tool registration for 'analyze_schema_relationships'. Instantiates SchemaMappingTool with SQL driver, fetches user schemas, calls the tool's analyze_schema_relationships method, and formats the text response.

  @mcp.tool(description="Analyze schema relationships and dependencies with visual representation")
  async def analyze_schema_relationships() -> ResponseType:
      """Analyze inter-schema dependencies and relationships with visual representation data."""
      try:
          sql_driver = await get_sql_driver()
          mapping_tool = SchemaMappingTool(sql_driver)
  
          # Get user schemas
          user_schemas_query = """
              SELECT schema_name
              FROM information_schema.schemata
              WHERE schema_name NOT IN ('information_schema', 'pg_catalog', 'pg_toast')
              AND schema_name NOT LIKE 'pg_temp_%'
              AND schema_name NOT LIKE 'pg_toast_temp_%'
              ORDER BY schema_name
          """
  
          rows = await sql_driver.execute_query(user_schemas_query)
          user_schemas = [row.cells["schema_name"] for row in rows] if rows else []
  
          # Analyze schema relationships
          result = await mapping_tool.analyze_schema_relationships(user_schemas)
  
          return format_text_response(result)
  
      except Exception as e:
          logger.error(f"Error analyzing schema relationships: {e}")
          return format_error_response(str(e))

• src/postgres_mcp_pro_plus/schema_mapping.py:79-105 (handler)

  Core handler in SchemaMappingTool class that performs the schema relationship analysis: resets state, initializes SchemaNode instances, analyzes schemas and tables via helpers, builds mappings, generates results, and formats as text.

  async def analyze_schema_relationships(self, schemas: list[str]) -> str:
      """Analyze relationships between schemas and generate mapping data."""
      try:
          # Reset state
          self.schema_nodes = {}
          self.table_nodes = {}
          self.cross_schema_relationships = []
          self.intra_schema_relationships = []
  
          # Initialize schema nodes
          for schema in schemas:
              self.schema_nodes[schema] = SchemaNode(name=schema)
  
          # Analyze each schema
          for schema in schemas:
              await self._analyze_schema(schema)
  
          # Build relationship mappings
          await self._build_relationship_mappings()
  
          # Generate analysis results
          result = await self._generate_analysis_results()
          return self._format_as_text(result)
  
      except Exception as e:
          logger.error(f"Error analyzing schema relationships: {e}")
          return f"Error analyzing schema relationships: {e}"

• src/postgres_mcp_pro_plus/schema_mapping.py:200-235 (helper)

  Helper method that aggregates analysis from schema dependencies, table dependencies, relationship patterns, visual data, and recommendations into a comprehensive result dictionary.

  async def _generate_analysis_results(self) -> dict[str, Any]:
      """Generate comprehensive analysis results."""
      try:
          # Schema analysis
          schema_analysis = self._analyze_schema_dependencies()
  
          # Table analysis
          table_analysis = self._analyze_table_dependencies()
  
          # Relationship patterns
          relationship_patterns = self._analyze_relationship_patterns()
  
          # Visual representation data
          visual_data = self._generate_visual_representation()
  
          # Recommendations
          recommendations = self._generate_recommendations()
  
          return {
              "schema_analysis": schema_analysis,
              "table_analysis": table_analysis,
              "relationship_patterns": relationship_patterns,
              "visual_representation": visual_data,
              "recommendations": recommendations,
              "summary": {
                  "total_schemas": len(self.schema_nodes),
                  "total_tables": len(self.table_nodes),
                  "cross_schema_relationships": len(self.cross_schema_relationships),
                  "intra_schema_relationships": len(self.intra_schema_relationships),
              },
          }
  
      except Exception as e:
          logger.error(f"Error generating analysis results: {e}")
          raise

• src/postgres_mcp_pro_plus/schema_mapping.py:18-39 (schema)

  Dataclass defining SchemaNode with fields and properties for dependency and isolation scoring, used throughout the analysis.

  @dataclass
  class SchemaNode:
      """Represents a schema node in the dependency graph."""
  
      name: str
      table_count: int = 0
      total_size_bytes: int = 0
      total_rows: int = 0
      outgoing_references: set[str] = field(default_factory=set)
      incoming_references: set[str] = field(default_factory=set)
      self_references: int = 0
  
      @property
      def dependency_score(self) -> float:
          """Calculate dependency score based on incoming and outgoing references."""
          return len(self.incoming_references) * 2 + len(self.outgoing_references)
  
      @property
      def isolation_score(self) -> float:
          """Calculate isolation score (lower is more isolated)."""
          return len(self.incoming_references) + len(self.outgoing_references)

• src/postgres_mcp_pro_plus/schema_mapping.py:41-67 (schema)

  Dataclass defining TableNode with fields and properties for connection count, hub status, and isolation, used in table-level analysis.

  @dataclass
  class TableNode:
      """Represents a table node in the dependency graph."""
  
      schema: str
      name: str
      qualified_name: str
      size_bytes: int = 0
      row_count: int = 0
      outgoing_fks: list[str] = field(default_factory=list)
      incoming_fks: list[str] = field(default_factory=list)
  
      @property
      def connection_count(self) -> int:
          """Total number of connections (incoming + outgoing)."""
          return len(self.outgoing_fks) + len(self.incoming_fks)
  
      @property
      def is_hub(self) -> bool:
          """Check if table is a hub (has many incoming references)."""
          return len(self.incoming_fks) >= 3
  
      @property
      def is_isolated(self) -> bool:
          """Check if table has no foreign key relationships."""
          return len(self.outgoing_fks) == 0 and len(self.incoming_fks) == 0