Schema Introspection

class AnyType:
    """
    Base type descriptor for EdgeDB types.
    
    Provides common attributes for all EdgeDB type descriptors.
    """
    
    def __init__(self, desc_id: UUID, name: Optional[str] = None):
        """
        Create type descriptor.
        
        Parameters:
        - desc_id: Unique identifier for the type
        - name: Optional type name
        """
    
    @property
    def desc_id(self) -> UUID:
        """Unique identifier for this type."""
    
    @property
    def name(self) -> Optional[str]:
        """Type name if available."""

class Element:
    """
    Schema element descriptor.
    
    Represents a property, link, or link property in the schema.
    """
    
    def __init__(
        self,
        type: AnyType,
        cardinality: Cardinality,
        is_implicit: bool = False,
        kind: ElementKind = ElementKind.PROPERTY
    ):
        """
        Create element descriptor.
        
        Parameters:
        - type: Element type descriptor
        - cardinality: Element cardinality
        - is_implicit: Whether element is implicit
        - kind: Kind of schema element
        """
    
    @property
    def type(self) -> AnyType:
        """Element type."""
    
    @property
    def cardinality(self) -> Cardinality:
        """Element cardinality."""
    
    @property
    def is_implicit(self) -> bool:
        """Whether element is implicit."""
    
    @property
    def kind(self) -> ElementKind:
        """Schema element kind."""

class SequenceType(AnyType):
    """
    Base descriptor for sequence types (arrays, sets, etc.).
    
    Extends AnyType with element type information.
    """
    
    def __init__(self, desc_id: UUID, name: Optional[str], element_type: AnyType):
        """
        Create sequence type descriptor.
        
        Parameters:
        - desc_id: Type identifier
        - name: Type name
        - element_type: Type of sequence elements
        """
    
    @property
    def element_type(self) -> AnyType:
        """Type of sequence elements."""

class SetType(SequenceType):
    """
    Set type descriptor.
    
    Represents EdgeDB set types.
    """

class DescribeContext:
    """
    Context for schema introspection operations.
    
    Controls what information to include in introspection results.
    """
    
    def __init__(
        self,
        query: str,
        state: Optional[State] = None,
        inject_type_names: bool = False,
        output_format: OutputFormat = OutputFormat.BINARY,
        expect_one: bool = False
    ):
        """
        Create describe context.
        
        Parameters:
        - query: Query to introspect
        - state: Client state for introspection
        - inject_type_names: Whether to inject type names
        - output_format: Output format for introspection
        - expect_one: Whether to expect single result
        """
    
    @property
    def query(self) -> str:
        """Query being introspected."""
    
    @property
    def state(self) -> Optional[State]:
        """Client state for introspection."""
    
    @property
    def inject_type_names(self) -> bool:
        """Whether to inject type names."""
    
    @property
    def output_format(self) -> OutputFormat:
        """Output format for results."""
    
    @property
    def expect_one(self) -> bool:
        """Whether single result expected."""

class DescribeResult:
    """
    Result of schema introspection operation.
    
    Contains type information and metadata about a query.
    """
    
    def __init__(
        self,
        input_type: Optional[AnyType] = None,
        output_type: Optional[AnyType] = None,
        output_cardinality: Cardinality = Cardinality.MANY,
        capabilities: int = 0
    ):
        """
        Create describe result.
        
        Parameters:
        - input_type: Input parameter type descriptor
        - output_type: Output result type descriptor
        - output_cardinality: Output cardinality
        - capabilities: Required capabilities for query
        """
    
    @property
    def input_type(self) -> Optional[AnyType]:
        """Input parameter type."""
    
    @property
    def output_type(self) -> Optional[AnyType]:
        """Output result type."""
    
    @property
    def output_cardinality(self) -> Cardinality:
        """Output cardinality."""
    
    @property
    def capabilities(self) -> int:
        """Required capabilities bitmask."""

def introspect_type(client: Union[Client, AsyncIOClient], type_name: str) -> AnyType:
    """
    Introspect a specific type in the database schema.
    
    Parameters:
    - client: EdgeDB client instance
    - type_name: Name of type to introspect
    
    Returns:
    Type descriptor for the specified type
    """

def describe_query(
    client: Union[Client, AsyncIOClient],
    query: str,
    *args,
    **kwargs
) -> DescribeResult:
    """
    Describe a query's input and output types.
    
    Parameters:
    - client: EdgeDB client instance
    - query: EdgeQL query to describe
    - *args: Query arguments
    - **kwargs: Named query arguments
    
    Returns:
    Description of query types and cardinality
    """

def get_schema_version(client: Union[Client, AsyncIOClient]) -> str:
    """
    Get the current schema version.
    
    Parameters:
    - client: EdgeDB client instance
    
    Returns:
    Schema version string
    """

import edgedb

client = edgedb.create_client()

# Describe a simple query
result = client.describe_query("SELECT User { name, email }")

print(f"Output cardinality: {result.output_cardinality}")
print(f"Output type: {result.output_type.name}")
print(f"Capabilities required: {result.capabilities}")

# Describe parameterized query
result = client.describe_query(
    "SELECT User { name } FILTER .id = <uuid>$user_id",
    user_id="123e4567-e89b-12d3-a456-426614174000"
)

if result.input_type:
    print(f"Input type: {result.input_type.name}")

import edgedb

client = edgedb.create_client()

# Describe complex nested query
query = """
    SELECT Article {
        title,
        content,
        author: { name, email },
        tags,
        comments: {
            content,
            author: { name },
            replies: { content, author: { name } }
        }
    }
    FILTER .published = true
"""

result = client.describe_query(query)

def analyze_type(type_desc, level=0):
    """Recursively analyze type structure."""
    indent = "  " * level
    print(f"{indent}Type: {type_desc.name or 'anonymous'}")
    
    if hasattr(type_desc, 'element_type'):
        print(f"{indent}  Element type:")
        analyze_type(type_desc.element_type, level + 2)
    
    if hasattr(type_desc, 'fields'):
        print(f"{indent}  Fields:")
        for field_name, field_desc in type_desc.fields.items():
            print(f"{indent}    {field_name}:")
            analyze_type(field_desc.type, level + 3)

if result.output_type:
    analyze_type(result.output_type)

import edgedb

client = edgedb.create_client()

# Get current schema version
version = client.get_schema_version()
print(f"Current schema version: {version}")

# Store version for comparison
stored_version = version

# ... perform schema migrations ...

# Check if schema changed
new_version = client.get_schema_version()
if new_version != stored_version:
    print("Schema has been updated")
    print(f"Old version: {stored_version}")
    print(f"New version: {new_version}")

import edgedb

client = edgedb.create_client()

def analyze_query_capabilities(query: str):
    """Analyze what capabilities a query requires."""
    result = client.describe_query(query)
    
    capabilities = result.capabilities
    required_caps = []
    
    # Check specific capability flags
    if capabilities & edgedb.Capability.MODIFICATIONS:
        required_caps.append("MODIFICATIONS")
    if capabilities & edgedb.Capability.SESSION_CONFIG:
        required_caps.append("SESSION_CONFIG")
    if capabilities & edgedb.Capability.TRANSACTION:
        required_caps.append("TRANSACTION")
    if capabilities & edgedb.Capability.DDL:
        required_caps.append("DDL")
    
    print(f"Query: {query}")
    print(f"Required capabilities: {', '.join(required_caps) or 'None'}")
    return required_caps

# Analyze different types of queries
analyze_query_capabilities("SELECT User { name }")
analyze_query_capabilities("INSERT User { name := 'Alice' }")
analyze_query_capabilities("CREATE TYPE NewType { name: str }")
analyze_query_capabilities("CONFIGURE SESSION SET query_execution_timeout := '30s'")

import edgedb

client = edgedb.create_client()

def explore_object_type(type_name: str):
    """Explore the structure of an object type."""
    
    query = f"""
        SELECT schema::ObjectType {{
            name,
            properties: {{
                name,
                target: {{ name }},
                cardinality,
                required
            }},
            links: {{
                name,
                target: {{ name }},
                cardinality,
                required
            }}
        }}
        FILTER .name = '{type_name}'
    """
    
    type_info = client.query_single(query)
    
    if not type_info:
        print(f"Type '{type_name}' not found")
        return
    
    print(f"Object Type: {type_info.name}")
    
    print("\nProperties:")
    for prop in type_info.properties:
        cardinality = "REQUIRED" if prop.required else "OPTIONAL"
        print(f"  {prop.name}: {prop.target.name} [{cardinality}, {prop.cardinality}]")
    
    print("\nLinks:")
    for link in type_info.links:
        cardinality = "REQUIRED" if link.required else "OPTIONAL"
        print(f"  {link.name} -> {link.target.name} [{cardinality}, {link.cardinality}]")

# Explore specific types
explore_object_type("User")
explore_object_type("Article")

import edgedb

def validate_schema_compatibility(client, expected_types):
    """Validate that schema contains expected types and structure."""
    
    issues = []
    
    for type_name in expected_types:
        try:
            # Check if type exists
            query = f"SELECT schema::ObjectType FILTER .name = '{type_name}'"
            type_exists = client.query_single(query)
            
            if not type_exists:
                issues.append(f"Missing type: {type_name}")
                continue
            
            # Validate type structure
            result = client.describe_query(f"SELECT {type_name} {{ * }}")
            
            if not result.output_type:
                issues.append(f"Cannot describe type: {type_name}")
            
        except edgedb.EdgeDBError as e:
            issues.append(f"Error checking type {type_name}: {e}")
    
    return issues

# Usage
client = edgedb.create_client()
expected_types = ["User", "Article", "Comment", "Tag"]
schema_issues = validate_schema_compatibility(client, expected_types)

if schema_issues:
    print("Schema validation issues:")
    for issue in schema_issues:
        print(f"  - {issue}")
else:
    print("Schema validation passed")

import edgedb
from typing import Dict, Any

def analyze_query_performance(client, queries: Dict[str, str]):
    """Analyze queries for performance characteristics."""
    
    analysis_results = {}
    
    for name, query in queries.items():
        try:
            result = client.describe_query(query)
            
            analysis = {
                'cardinality': result.output_cardinality.value,
                'capabilities': result.capabilities,
                'has_filters': 'FILTER' in query.upper(),
                'has_order': 'ORDER BY' in query.upper(),
                'has_limit': 'LIMIT' in query.upper(),
                'complexity_score': calculate_complexity(query)
            }
            
            analysis_results[name] = analysis
            
        except edgedb.EdgeDBError as e:
            analysis_results[name] = {'error': str(e)}
    
    return analysis_results

def calculate_complexity(query: str) -> int:
    """Simple query complexity score."""
    score = 0
    score += query.count('SELECT') * 1
    score += query.count('FILTER') * 2
    score += query.count('ORDER BY') * 1
    score += query.count('GROUP BY') * 3
    score += query.count('{') * 1  # Nested selections
    return score

# Usage
queries = {
    'simple_user_query': "SELECT User { name, email }",
    'complex_article_query': """
        SELECT Article {
            title,
            author: { name },
            comments: { content, author: { name } },
            tag_count := count(.tags)
        }
        FILTER .published = true
        ORDER BY .created_at DESC
        LIMIT 10
    """,
    'aggregation_query': """
        SELECT User {
            name,
            article_count := count(.articles),
            avg_comment_count := math::mean(.articles.comment_count)
        }
        GROUP BY .department
    """
}

client = edgedb.create_client()
analysis = analyze_query_performance(client, queries)

for query_name, results in analysis.items():
    print(f"\n{query_name}:")
    if 'error' in results:
        print(f"  Error: {results['error']}")
    else:
        print(f"  Cardinality: {results['cardinality']}")
        print(f"  Complexity Score: {results['complexity_score']}")
        print(f"  Has Filters: {results['has_filters']}")
        print(f"  Has Ordering: {results['has_order']}")

import edgedb

def build_filtered_query(client, type_name, filters=None, fields=None):
    """Build query dynamically based on type introspection."""
    
    # Get type information
    type_query = f"""
        SELECT schema::ObjectType {{
            properties: {{ name, target: {{ name }} }},
            links: {{ name, target: {{ name }} }}
        }}
        FILTER .name = '{type_name}'
    """
    
    type_info = client.query_single(type_query)
    if not type_info:
        raise ValueError(f"Type {type_name} not found")
    
    # Build field selection
    if fields is None:
        # Select all scalar properties by default
        fields = [prop.name for prop in type_info.properties 
                 if prop.target.name in ['std::str', 'std::int64', 'std::bool', 'std::datetime']]
    
    field_selection = "{ " + ", ".join(fields) + " }"
    
    # Build base query
    query = f"SELECT {type_name} {field_selection}"
    
    # Add filters if provided
    if filters:
        filter_parts = []
        for field, value in filters.items():
            if isinstance(value, str):
                filter_parts.append(f".{field} = '{value}'")
            else:
                filter_parts.append(f".{field} = {value}")
        
        if filter_parts:
            query += " FILTER " + " AND ".join(filter_parts)
    
    return query

# Usage
client = edgedb.create_client()

# Build query dynamically
query = build_filtered_query(
    client,
    "User",
    filters={"active": True, "role": "admin"},
    fields=["name", "email", "created_at"]
)

print(f"Generated query: {query}")
results = client.query(query)