🔍 Code Extractor

Flask route handler that retrieves and formats detailed database schema information from a discovered schema stored in the Flask app object, returning it as JSON for visualization purposes.

Purpose

This endpoint serves as an API to expose comprehensive database schema metadata including tables, columns, relationships, and quality metrics. It handles both TableInfo objects and dictionary representations of schema data, making it flexible for different data sources. The endpoint is designed to support database schema visualization tools by providing structured information about database structure, data types, constraints, and relationships.

Source Code

def get_database_schema_viewer():
    """Get detailed database schema information for visualization"""
    try:
        if hasattr(app, 'discovered_schema') and app.discovered_schema:
            schema_data = {
                'database_name': app.discovered_schema.database_name,
                'server_name': app.discovered_schema.server_name,
                'discovery_timestamp': app.discovered_schema.discovery_timestamp,
                'total_tables': app.discovered_schema.total_tables,
                'total_columns': app.discovered_schema.total_columns,
                'total_relationships': len(app.discovered_schema.relationships),
                'tables': [],
                'relationships': app.discovered_schema.relationships
            }
            
            for table in app.discovered_schema.tables:
                # Handle both TableInfo objects and dictionaries
                if isinstance(table, dict):
                    # Table is a dictionary (from direct JSON loading)
                    table_data = {
                        'name': table.get('name'),
                        'schema': table.get('schema'),
                        'full_name': f"{table.get('schema', '')}.{table.get('name', '')}",
                        'table_type': table.get('table_type'),
                        'row_count': table.get('row_count'),
                        'description': table.get('description') or 'No description available',
                        'quality_score': table.get('data_quality_score'),
                        'column_count': len(table.get('columns', [])),
                        'columns': []
                    }
                    
                    # Handle columns (they are also dictionaries in this case)
                    for column in table.get('columns', []):
                        column_data = {
                            'name': column.get('name'),
                            'data_type': column.get('data_type'),
                            'is_nullable': column.get('is_nullable'),
                            'is_primary_key': column.get('is_primary_key'),
                            'default_value': column.get('default_value'),
                            'character_maximum_length': column.get('max_length'),  # Use correct field name from cache
                            'numeric_precision': column.get('precision'),  # Use correct field name from cache
                            'numeric_scale': column.get('scale')  # Use correct field name from cache
                        }
                        table_data['columns'].append(column_data)
                else:
                    # Table is a TableInfo object (expected case)
                    table_data = {
                        'name': table.name,
                        'schema': table.schema,
                        'full_name': f"{table.schema}.{table.name}",
                        'table_type': table.table_type,
                        'row_count': table.row_count,
                        'description': table.description or 'No description available',
                        'quality_score': getattr(table, 'data_quality_score', None),
                        'column_count': len(table.columns),
                        'columns': []
                    }
                    
                    # Handle columns for TableInfo objects
                    for column in table.columns:
                        # Columns are stored as dictionaries in both cases
                        column_data = {
                            'name': column['name'],
                            'data_type': column['data_type'],
                            'is_nullable': column['is_nullable'],
                            'is_primary_key': column['is_primary_key'],
                            'default_value': column['default_value'],
                            'character_maximum_length': column.get('max_length'),  # Use correct field name
                            'numeric_precision': column.get('precision'),  # Use correct field name
                            'numeric_scale': column.get('scale')  # Use correct field name
                        }
                        table_data['columns'].append(column_data)
                
                schema_data['tables'].append(table_data)
            
            return jsonify({
                'success': True,
                'schema': schema_data
            }), 200
        else:
            return jsonify({
                'success': False,
                'error': 'Database schema not available. Please ensure the application has connected to the database.'
            }), 404
            
    except Exception as e:
        logger.error(f"Error retrieving database schema for viewer: {str(e)}")
        return jsonify({
            'success': False,
            'error': str(e)
        }), 500

Return Value

Returns a Flask JSON response tuple containing (response_body, status_code). On success (200): {'success': True, 'schema': {...}} where schema contains database_name, server_name, discovery_timestamp, total_tables, total_columns, total_relationships, tables array (with columns, types, constraints), and relationships array. On not found (404): {'success': False, 'error': 'Database schema not available...'} when schema hasn't been discovered. On error (500): {'success': False, 'error': error_message} for exceptions.

Dependencies

• flask
• logging

Required Imports

from flask import jsonify
import logging

Usage Example

# Assuming Flask app is set up with discovered schema
from flask import Flask, jsonify
import logging

logger = logging.getLogger(__name__)
app = Flask(__name__)

# Mock schema data structure
class MockSchema:
    def __init__(self):
        self.database_name = 'my_database'
        self.server_name = 'localhost'
        self.discovery_timestamp = '2024-01-01T12:00:00'
        self.total_tables = 2
        self.total_columns = 10
        self.relationships = [{'from': 'users.id', 'to': 'orders.user_id'}]
        self.tables = [
            {
                'name': 'users',
                'schema': 'dbo',
                'table_type': 'BASE TABLE',
                'row_count': 1000,
                'description': 'User accounts',
                'data_quality_score': 0.95,
                'columns': [
                    {
                        'name': 'id',
                        'data_type': 'int',
                        'is_nullable': False,
                        'is_primary_key': True,
                        'default_value': None,
                        'max_length': None,
                        'precision': 10,
                        'scale': 0
                    }
                ]
            }
        ]

app.discovered_schema = MockSchema()

@app.route('/database_schema_viewer', methods=['GET'])
def get_database_schema_viewer():
    # Function implementation here
    pass

# Make GET request to /database_schema_viewer
# Returns JSON with complete schema information

Best Practices

• Ensure app.discovered_schema is populated before calling this endpoint, typically through a schema discovery process during application initialization
• The function handles both TableInfo objects and dictionary representations, making it resilient to different data source formats
• Error handling is comprehensive with specific status codes (404 for missing schema, 500 for exceptions)
• Column field names are mapped correctly between cache format (max_length, precision, scale) and output format (character_maximum_length, numeric_precision, numeric_scale)
• Consider implementing caching for this endpoint if schema data is large and doesn't change frequently
• The endpoint returns all schema data in a single response; for very large databases, consider implementing pagination or filtering
• Ensure proper logging configuration to capture errors for debugging
• The hasattr check prevents AttributeError if discovered_schema is not set on the app object

Tags

Similar Components

AI-powered semantic similarity - components with related functionality:

• function get_database_schema 88.9% similar

  Flask route handler that retrieves and returns comprehensive database schema information, including tables, columns, relationships, and statistics.

  From: /tf/active/vicechatdev/full_smartstat/app.py

• function database_schema 84.9% similar

  Flask route handler that renders the database schema viewer page template.

  From: /tf/active/vicechatdev/full_smartstat/app.py

• function get_database_tables_columns 75.6% similar

  Flask route handler that retrieves database schema information including tables, columns, and relationships, filtered and sorted by relevance for data analysis workflows.

  From: /tf/active/vicechatdev/full_smartstat/app.py

• function refresh_database_schema 68.0% similar

  Flask route endpoint that forces a refresh of the database schema by invoking the schema discovery service with LLM-generated descriptions.

  From: /tf/active/vicechatdev/full_smartstat/app.py

• function load_sql_data 56.6% similar

  Flask route handler that loads data from a SQL database using a provided connection string and query, creating a data source for a specific analysis session.

  From: /tf/active/vicechatdev/full_smartstat/app.py