Schema Validation and Processing

class Schemas:
    """Manage JSON schemas for API validation and documentation."""
    
    def __init__(self, discovery_doc):
        """
        Initialize schema manager with discovery document.
        
        Args:
            discovery_doc (dict): Complete API discovery document containing schemas
        """
    
    def get(self, name, default=None):
        """
        Get a schema definition by name.
        
        Args:
            name (str): Name of the schema to retrieve
            default (object, optional): Default value if schema not found
            
        Returns:
            dict: Schema definition, or default if not found
        """
    
    def prettyPrintByName(self, name):
        """
        Pretty print a schema definition by name.
        
        Args:
            name (str): Name of the schema to pretty print
            
        Returns:
            str: Formatted string representation of the schema
        """
    
    def prettyPrintSchema(self, val):
        """
        Pretty print a schema definition.
        
        Args:
            val (dict): Schema definition to pretty print
            
        Returns:
            str: Formatted string representation of the schema
        """
    
    def _prettyPrintByName(self, name, seen):
        """
        Internal method for pretty printing schema by name with circular reference tracking.
        
        Args:
            name (str): Schema name to print
            seen (set): Set of already processed schemas to prevent infinite recursion
            
        Returns:
            str: Formatted schema representation
        """
    
    def _prettyPrintSchema(self, val, seen):
        """
        Internal method for pretty printing schema with circular reference tracking.
        
        Args:
            val (dict): Schema definition to print
            seen (set): Set of already processed schemas to prevent infinite recursion
            
        Returns:
            str: Formatted schema representation
        """

from googleapiclient import discovery
from googleapiclient.schema import Schemas

# Build service and get discovery document
service = discovery.build('gmail', 'v1', credentials=credentials)
discovery_doc = service._rootDesc  # Access discovery document

# Create schema manager
schemas = Schemas(discovery_doc)

# Get a specific schema
message_schema = schemas.get('Message')
if message_schema:
    print("Message schema found:")
    print(schemas.prettyPrintSchema(message_schema))
else:
    print("Message schema not found")

# Pretty print schema by name
thread_schema_str = schemas.prettyPrintByName('Thread')
print("Thread schema:")
print(thread_schema_str)

from googleapiclient.schema import Schemas
import json

def explore_api_schemas(service_name, version):
    """Explore all schemas in an API."""
    
    service = discovery.build(service_name, version, credentials=credentials)
    discovery_doc = service._rootDesc
    
    schemas = Schemas(discovery_doc)
    
    # Get all available schemas
    schema_names = discovery_doc.get('schemas', {}).keys()
    
    print(f"Available schemas in {service_name} {version}:")
    for name in sorted(schema_names):
        schema = schemas.get(name)
        if schema:
            print(f"\n{name}:")
            print(schemas.prettyPrintByName(name))
            print("-" * 50)

# Explore Gmail API schemas
explore_api_schemas('gmail', 'v1')

from googleapiclient.schema import Schemas
import jsonschema

class SchemaValidator:
    """Validate data against API schemas."""
    
    def __init__(self, discovery_doc):
        self.schemas = Schemas(discovery_doc)
        self.discovery_doc = discovery_doc
    
    def validate_data(self, schema_name, data):
        """
        Validate data against a named schema.
        
        Args:
            schema_name (str): Name of the schema to validate against
            data (dict): Data to validate
            
        Returns:
            tuple: (is_valid, errors) - validation result and error list
        """
        schema_def = self.schemas.get(schema_name)
        if not schema_def:
            return False, [f"Schema '{schema_name}' not found"]
        
        try:
            # Convert Google API schema to JSON Schema format
            json_schema = self._convert_to_json_schema(schema_def)
            jsonschema.validate(data, json_schema)
            return True, []
        except jsonschema.ValidationError as e:
            return False, [str(e)]
        except Exception as e:
            return False, [f"Validation error: {e}"]
    
    def _convert_to_json_schema(self, api_schema):
        """Convert Google API schema to JSON Schema format."""
        # This is a simplified conversion - real implementation would be more complex
        json_schema = {
            "type": "object",
            "properties": {},
            "required": []
        }
        
        if "properties" in api_schema:
            for prop_name, prop_def in api_schema["properties"].items():
                json_schema["properties"][prop_name] = self._convert_property(prop_def)
                if prop_def.get("required", False):
                    json_schema["required"].append(prop_name)
        
        return json_schema
    
    def _convert_property(self, prop_def):
        """Convert a single property definition."""
        prop_schema = {}
        
        # Map Google API types to JSON Schema types
        type_mapping = {
            "string": "string",
            "integer": "integer", 
            "number": "number",
            "boolean": "boolean",
            "array": "array",
            "object": "object"
        }
        
        api_type = prop_def.get("type", "string")
        prop_schema["type"] = type_mapping.get(api_type, "string")
        
        if "description" in prop_def:
            prop_schema["description"] = prop_def["description"]
        
        if api_type == "array" and "items" in prop_def:
            prop_schema["items"] = self._convert_property(prop_def["items"])
        
        return prop_schema

# Usage
service = discovery.build('gmail', 'v1', credentials=credentials)
validator = SchemaValidator(service._rootDesc)

# Validate message data
message_data = {
    "id": "message123",
    "threadId": "thread456", 
    "labelIds": ["INBOX", "UNREAD"],
    "snippet": "This is a test message..."
}

is_valid, errors = validator.validate_data('Message', message_data)
if is_valid:
    print("Message data is valid")
else:
    print("Validation errors:")
    for error in errors:
        print(f"  - {error}")

from googleapiclient.schema import Schemas

class SchemaDocumentationGenerator:
    """Generate documentation from API schemas."""
    
    def __init__(self, discovery_doc):
        self.schemas = Schemas(discovery_doc)
        self.discovery_doc = discovery_doc
    
    def generate_markdown_docs(self, output_file):
        """Generate Markdown documentation for all schemas."""
        
        with open(output_file, 'w') as f:
            f.write("# API Schema Documentation\n\n")
            
            schema_names = self.discovery_doc.get('schemas', {}).keys()
            
            for name in sorted(schema_names):
                f.write(f"## {name}\n\n")
                
                schema = self.schemas.get(name)
                if schema and 'description' in schema:
                    f.write(f"{schema['description']}\n\n")
                
                # Pretty print the schema
                schema_str = self.schemas.prettyPrintByName(name)
                f.write("```\n")
                f.write(schema_str)
                f.write("\n```\n\n")
                
                # Add properties details
                if schema and 'properties' in schema:
                    f.write("### Properties\n\n")
                    for prop_name, prop_def in schema['properties'].items():
                        f.write(f"- **{prop_name}** ({prop_def.get('type', 'unknown')}): ")
                        f.write(f"{prop_def.get('description', 'No description')}\n")
                    f.write("\n")
    
    def get_schema_summary(self):
        """Get a summary of all schemas."""
        schema_names = self.discovery_doc.get('schemas', {}).keys()
        summary = {
            'total_schemas': len(schema_names),
            'schemas': {}
        }
        
        for name in schema_names:
            schema = self.schemas.get(name)
            if schema:
                summary['schemas'][name] = {
                    'description': schema.get('description', ''),
                    'property_count': len(schema.get('properties', {})),
                    'type': schema.get('type', 'object')
                }
        
        return summary

# Usage
service = discovery.build('gmail', 'v1', credentials=credentials)
doc_generator = SchemaDocumentationGenerator(service._rootDesc)

# Generate documentation
doc_generator.generate_markdown_docs('gmail_schemas.md')

# Get summary
summary = doc_generator.get_schema_summary()
print(f"Found {summary['total_schemas']} schemas")

for name, info in summary['schemas'].items():
    print(f"{name}: {info['property_count']} properties - {info['description'][:50]}...")

from googleapiclient.schema import Schemas

def compare_api_versions(service_name, version1, version2):
    """Compare schemas between two API versions."""
    
    # Build services for both versions
    service1 = discovery.build(service_name, version1, credentials=credentials)
    service2 = discovery.build(service_name, version2, credentials=credentials)
    
    schemas1 = Schemas(service1._rootDesc)
    schemas2 = Schemas(service2._rootDesc)
    
    # Get schema names from both versions
    names1 = set(service1._rootDesc.get('schemas', {}).keys())
    names2 = set(service2._rootDesc.get('schemas', {}).keys())
    
    # Find differences
    added_schemas = names2 - names1
    removed_schemas = names1 - names2
    common_schemas = names1 & names2
    
    print(f"Schema comparison: {service_name} {version1} vs {version2}")
    print(f"Added schemas: {len(added_schemas)}")
    for name in sorted(added_schemas):
        print(f"  + {name}")
    
    print(f"Removed schemas: {len(removed_schemas)}")
    for name in sorted(removed_schemas):
        print(f"  - {name}")
    
    print(f"Modified schemas:")
    for name in sorted(common_schemas):
        schema1 = schemas1.get(name)
        schema2 = schemas2.get(name)
        
        if schema1 != schema2:
            print(f"  ~ {name}")
            
            # Compare properties
            props1 = set(schema1.get('properties', {}).keys()) if schema1 else set()
            props2 = set(schema2.get('properties', {}).keys()) if schema2 else set()
            
            added_props = props2 - props1
            removed_props = props1 - props2
            
            if added_props:
                print(f"    Added properties: {', '.join(sorted(added_props))}")
            if removed_props:
                print(f"    Removed properties: {', '.join(sorted(removed_props))}")

# Compare Gmail API versions
compare_api_versions('gmail', 'v1', 'v1')  # Same version for demo

from googleapiclient.schema import Schemas

class InteractiveSchemaExplorer:
    """Interactive tool for exploring API schemas."""
    
    def __init__(self, service_name, version):
        self.service = discovery.build(service_name, version, credentials=credentials)
        self.schemas = Schemas(self.service._rootDesc)
        self.schema_names = list(self.service._rootDesc.get('schemas', {}).keys())
    
    def explore(self):
        """Start interactive exploration."""
        print(f"Schema Explorer - {len(self.schema_names)} schemas available")
        print("Commands: list, show <name>, search <term>, quit")
        
        while True:
            try:
                command = input("\nschema> ").strip().lower()
                
                if command == 'quit':
                    break
                elif command == 'list':
                    self._list_schemas()
                elif command.startswith('show '):
                    schema_name = command[5:].strip()
                    self._show_schema(schema_name)
                elif command.startswith('search '):
                    term = command[7:].strip()
                    self._search_schemas(term)
                else:
                    print("Unknown command. Use: list, show <name>, search <term>, quit")
                    
            except KeyboardInterrupt:
                break
            except Exception as e:
                print(f"Error: {e}")
    
    def _list_schemas(self):
        """List all available schemas."""
        print("Available schemas:")
        for i, name in enumerate(sorted(self.schema_names)):
            schema = self.schemas.get(name)
            desc = schema.get('description', 'No description') if schema else 'No description'
            print(f"  {i+1:2d}. {name} - {desc[:60]}...")
    
    def _show_schema(self, name):
        """Show detailed information about a schema."""
        if name not in self.schema_names:
            print(f"Schema '{name}' not found")
            return
        
        schema = self.schemas.get(name)
        if not schema:
            print(f"Could not retrieve schema '{name}'")
            return
        
        print(f"\nSchema: {name}")
        if 'description' in schema:
            print(f"Description: {schema['description']}")
        
        print("\nSchema definition:")
        print(self.schemas.prettyPrintByName(name))
    
    def _search_schemas(self, term):
        """Search for schemas containing the term."""
        matches = []
        for name in self.schema_names:
            schema = self.schemas.get(name)
            if schema:
                # Search in name and description
                if (term.lower() in name.lower() or 
                    term.lower() in schema.get('description', '').lower()):
                    matches.append(name)
        
        if matches:
            print(f"Found {len(matches)} matches:")
            for name in sorted(matches):
                schema = self.schemas.get(name)
                desc = schema.get('description', 'No description') if schema else 'No description'
                print(f"  {name} - {desc[:60]}...")
        else:
            print(f"No schemas found containing '{term}'")

# Usage
explorer = InteractiveSchemaExplorer('gmail', 'v1')
# explorer.explore()  # Uncomment for interactive use