Cache and Utilities

def call_linear_api(query: str | Dict[str, Any], api_key: Optional[str] = None) -> Dict[str, Any]:
    """
    Direct low-level interface to Linear's GraphQL API.

    Args:
        query: GraphQL query string or dictionary with query/variables
        api_key: Optional API key (falls back to LINEAR_API_KEY env var)

    Returns:
        API response data dictionary

    Raises:
        ValueError: For authentication or API errors

    Dependencies:
        requests, os
    """

from linear_api.utils import call_linear_api

# Simple query with query string
response = call_linear_api("""
    query {
        viewer {
            id
            name
            email
        }
    }
""")
print(f"Current user: {response['data']['viewer']['name']}")

# Query with variables using dictionary format
query = {
    "query": """
        query GetIssue($id: String!) {
            issue(id: $id) {
                id
                title
                state { name }
                assignee { name }
            }
        }
    """,
    "variables": {
        "id": "issue-id"
    }
}
response = call_linear_api(query)
issue_data = response['data']['issue']

def process_issue_data(data: Dict[str, Any]) -> LinearIssue:
    """
    Transform raw Linear API issue data into structured LinearIssue objects.

    Key Processing:
    - Converts nested objects (state, team, assignee, project, labels, attachments)
    - Handles datetime field parsing (ISO format → Python datetime)
    - Processes pagination nodes for collections
    - Adds default values for missing project fields
    - Handles parent issue relationships

    Args:
        data: Raw issue data from Linear API

    Returns:
        LinearIssue object with processed data

    Dependencies:
        LinearIssue, LinearUser, LinearState, LinearLabel, LinearProject, LinearTeam, LinearAttachment
    """

def process_project_data(data: Dict[str, Any]) -> LinearProject:
    """
    Transform raw Linear API project data into structured LinearProject objects.

    Key Processing:
    - Parses datetime fields from ISO strings
    - Handles special TimelessDate fields (startDate, targetDate)
    - Preserves project-specific data structures

    Args:
        data: Raw project data from Linear API

    Returns:
        LinearProject object with processed data

    Dependencies:
        LinearProject, TimelessDate
    """

from linear_api.utils import process_issue_data, call_linear_api

# Get raw issue data and process it
raw_response = call_linear_api("""
    query {
        issue(id: "issue-id") {
            id
            title
            state { id name type color }
            assignee { id name email }
            team { id name key }
            createdAt
            updatedAt
        }
    }
""")

# Process into structured model
issue = process_issue_data(raw_response['data']['issue'])
print(f"Processed issue: {issue.title}")
print(f"State: {issue.state.name}")
print(f"Assignee: {issue.assignee.name if issue.assignee else 'Unassigned'}")

def enrich_with_client(func):
    """
    Decorator that recursively attaches client references to returned model objects.

    Key Features:
    - Prevents infinite recursion by checking direct attributes only
    - Handles nested objects, lists, and dictionaries
    - Enables method chaining on returned models

    Usage:
        Applied to manager methods that return LinearModel instances
    """

from linear_api.utils import enrich_with_client

# This decorator is typically used internally by managers
@enrich_with_client
def custom_get_issue(client, issue_id: str):
    raw_data = client.call_api(f"query {{ issue(id: \"{issue_id}\") {{ ... }} }}")
    return process_issue_data(raw_data['data']['issue'])

# The returned issue will have client reference for dynamic properties
issue = custom_get_issue(client, "issue-id")
comments = issue.comments  # This works because of client enrichment

class CacheManager:
    """
    Manages caching for API responses with TTL support.
    """

    def __init__(self, enabled: bool = True, default_ttl: int = 3600):
        """
        Initialize cache manager.

        Args:
            enabled: Whether caching is enabled
            default_ttl: Default time-to-live in seconds
        """

    def get(self, cache_name: str, key: Any) -> Optional[Any]:
        """
        Get a value from the cache with expiration checking.

        Args:
            cache_name: Name of the cache to access
            key: Cache key to retrieve

        Returns:
            Cached value or None if not found/expired
        """

    def set(self, cache_name: str, key: Any, value: Any, ttl: Optional[int] = None) -> None:
        """
        Set a value in the cache with optional time-to-live.

        Args:
            cache_name: Name of the cache to store in
            key: Cache key
            value: Value to cache
            ttl: Optional TTL override (uses default if None)
        """

    def cached(self, cache_name: str, key_fn: Callable = lambda *args, **kwargs: str(args) + str(kwargs)):
        """
        Decorator for caching function results with customizable key generation.

        Args:
            cache_name: Name of cache to use
            key_fn: Function to generate cache keys from function arguments

        Returns:
            Decorator function
        """

    def clear(self, cache_name: Optional[str] = None) -> None:
        """
        Clear a specific cache or all caches.

        Args:
            cache_name: Cache to clear, or None for all caches
        """

    def invalidate(self, cache_name: str, key: Any) -> None:
        """
        Invalidate a specific cache entry.

        Args:
            cache_name: Name of the cache
            key: Key to invalidate
        """

    def enable(self) -> None:
        """Enable caching."""

    def disable(self) -> None:
        """Disable caching."""

    def get_cache_size(self, cache_name: Optional[str] = None) -> int:
        """
        Get the number of entries in a cache or all caches.

        Args:
            cache_name: Specific cache name, or None for total

        Returns:
            Number of cache entries
        """

    @property
    def enabled(self) -> bool:
        """Check or set whether caching is enabled."""

    @property
    def stats(self) -> Dict[str, Any]:
        """
        Get cache statistics including hit count, miss count, hit rate,
        and entry counts.

        Returns:
            Dictionary with cache statistics
        """

from linear_api import LinearClient

client = LinearClient()
cache = client.cache

# Check cache status
print(f"Cache enabled: {cache.enabled}")
print(f"Cache stats: {cache.stats}")

# Manual cache operations
cache.set("custom", "my-key", {"data": "value"}, ttl=300)  # 5 minutes
value = cache.get("custom", "my-key")

# Cache decorator for custom functions
@cache.cached("my-function")
def expensive_operation(param1, param2):
    # Simulate expensive operation
    time.sleep(1)
    return f"Result for {param1}, {param2}"

# First call is slow, subsequent calls are fast
result1 = expensive_operation("a", "b")  # Slow
result2 = expensive_operation("a", "b")  # Fast (cached)

# Cache management
cache.clear("my-function")  # Clear specific cache
cache.clear()  # Clear all caches
cache.disable()  # Disable caching
cache.enable()   # Re-enable caching

def monitor_cache_performance(client):
    """Monitor cache performance and hit rates."""
    stats = client.cache.stats

    print("Cache Performance Report:")
    print("=" * 40)
    print(f"Cache enabled: {client.cache.enabled}")
    print(f"Total hits: {stats.get('hits', 0)}")
    print(f"Total misses: {stats.get('misses', 0)}")
    print(f"Hit rate: {stats.get('hit_rate', 0):.2%}")

    # Per-cache statistics
    cache_sizes = {}
    for cache_name in ['issues', 'projects', 'teams', 'users']:
        size = client.cache.get_cache_size(cache_name)
        if size > 0:
            cache_sizes[cache_name] = size

    if cache_sizes:
        print("\nCache sizes:")
        for name, size in cache_sizes.items():
            print(f"  {name}: {size} entries")

    return stats

# Monitor performance
stats = monitor_cache_performance(client)

def validate_all_models(api_key: str) -> Dict[str, Dict[str, Any]]:
    """
    Validate all domain models against GraphQL schema.

    Args:
        api_key: Linear API key for schema access

    Returns:
        Dictionary with completeness metrics and field comparisons
    """

def validate_model(model_class: type[LinearModel], api_key: str) -> Dict[str, Any]:
    """
    Validate a single model class against GraphQL type.

    Args:
        model_class: LinearModel subclass to validate
        api_key: Linear API key for schema access

    Returns:
        Detailed field analysis and validation results
    """

def get_schema_for_type(type_name: str, api_key: str) -> Dict[str, Any]:
    """
    Use GraphQL introspection to get type schema.

    Args:
        type_name: GraphQL type name to introspect
        api_key: Linear API key

    Returns:
        Field definitions and metadata for the type
    """

def suggest_model_improvements(model_class: type[LinearModel], api_key: str) -> str:
    """
    Generate code suggestions for missing fields.

    Args:
        model_class: LinearModel subclass to analyze
        api_key: Linear API key

    Returns:
        Code suggestions with field definitions and Python type mappings
    """

def compare_fields(model_class: type[LinearModel], api_key: str) -> Tuple[Set[str], Set[str], Set[str]]:
    """
    Compare model fields vs GraphQL schema.

    Args:
        model_class: LinearModel subclass to compare
        api_key: Linear API key

    Returns:
        Tuple of (common_fields, missing_fields, extra_fields)
    """

from linear_api.schema_validator import validate_all_models, validate_model, suggest_model_improvements
from linear_api import LinearIssue

# Validate all models
all_results = validate_all_models(api_key="your-api-key")
for model_name, results in all_results.items():
    completeness = results.get('completeness', 0)
    print(f"{model_name}: {completeness:.1%} complete")

# Validate specific model
issue_validation = validate_model(LinearIssue, api_key="your-api-key")
print(f"Issue model completeness: {issue_validation['completeness']:.1%}")
print(f"Missing fields: {issue_validation['missing_fields']}")

# Get improvement suggestions
suggestions = suggest_model_improvements(LinearIssue, api_key="your-api-key")
print("Suggested improvements:")
print(suggestions)

def introspect_type(client, type_name: str) -> Dict[str, Any]:
    """
    Introspect GraphQL type structure.

    Args:
        client: LinearClient instance
        type_name: GraphQL type name to introspect

    Returns:
        Field information and metadata
    """

def get_field_names(client, type_name: str) -> List[str]:
    """
    Extract field names from GraphQL type.

    Args:
        client: LinearClient instance
        type_name: GraphQL type name

    Returns:
        Simplified field name list
    """

def print_type_fields(client, type_name: str) -> None:
    """
    Pretty-print type field information.

    Args:
        client: LinearClient instance
        type_name: GraphQL type name

    Includes:
        Field descriptions and type information
    """

from linear_api.utils.introspection_helper import introspect_type, print_type_fields

# Introspect a GraphQL type
issue_schema = introspect_type(client, "Issue")
print(f"Issue type has {len(issue_schema['fields'])} fields")

# Print detailed field information
print_type_fields(client, "Issue")

# Get field names only
field_names = get_field_names(client, "Project")
print(f"Project fields: {field_names}")

def get_issue_with_custom_fields(client, issue_id: str):
    """Get issue with specific field selection for performance."""
    query = """
        query GetIssueCustom($id: String!) {
            issue(id: $id) {
                id
                title
                description
                priority
                state { name type }
                assignee { name email }
                team { name key }
                labels(first: 10) {
                    nodes { name color }
                }
                comments(first: 5) {
                    nodes {
                        body
                        createdAt
                        user { name }
                    }
                }
            }
        }
    """

    response = client.execute_graphql(query, {"id": issue_id})
    return response['data']['issue']

# Use custom query
issue_data = get_issue_with_custom_fields(client, "issue-id")
print(f"Issue: {issue_data['title']}")
print(f"Comments: {len(issue_data['comments']['nodes'])}")

def bulk_process_issues(client, issue_ids: List[str]):
    """Process multiple issues efficiently with single query."""
    # Build query for multiple issues
    issue_queries = []
    for i, issue_id in enumerate(issue_ids):
        issue_queries.append(f"""
            issue{i}: issue(id: "{issue_id}") {{
                id
                title
                state {{ name }}
                assignee {{ name }}
                priority
            }}
        """)

    query = f"query {{ {' '.join(issue_queries)} }}"
    response = client.call_api(query)

    # Process results
    issues = []
    for i in range(len(issue_ids)):
        issue_data = response['data'][f'issue{i}']
        if issue_data:
            # Use process_issue_data for full model conversion
            issues.append(process_issue_data(issue_data))

    return issues

# Process multiple issues in single API call
issue_ids = ["issue-1", "issue-2", "issue-3"]
issues = bulk_process_issues(client, issue_ids)

def safe_api_call(client, query, variables=None, retries=3):
    """Make API call with retry logic and error handling."""
    import time

    for attempt in range(retries):
        try:
            if variables:
                return client.execute_graphql(query, variables)
            else:
                return client.call_api(query)
        except ValueError as e:
            if "rate limit" in str(e).lower() and attempt < retries - 1:
                # Exponential backoff for rate limiting
                wait_time = 2 ** attempt
                print(f"Rate limited, waiting {wait_time}s before retry...")
                time.sleep(wait_time)
                continue
            else:
                raise

    raise ValueError(f"Failed after {retries} attempts")

# Use with error handling
try:
    response = safe_api_call(client, "query { viewer { name } }")
    print(f"User: {response['data']['viewer']['name']}")
except ValueError as e:
    print(f"API call failed: {e}")

import time
from contextlib import contextmanager

@contextmanager
def profile_api_calls(client):
    """Profile API call performance and cache effectiveness."""
    start_stats = client.cache.stats.copy()
    start_time = time.time()

    yield

    end_time = time.time()
    end_stats = client.cache.stats

    duration = end_time - start_time
    cache_hits = end_stats.get('hits', 0) - start_stats.get('hits', 0)
    cache_misses = end_stats.get('misses', 0) - start_stats.get('misses', 0)
    total_requests = cache_hits + cache_misses

    print(f"Performance Profile:")
    print(f"  Duration: {duration:.2f}s")
    print(f"  Total requests: {total_requests}")
    print(f"  Cache hits: {cache_hits}")
    print(f"  Cache misses: {cache_misses}")
    if total_requests > 0:
        print(f"  Hit rate: {cache_hits/total_requests:.1%}")

# Profile a set of operations
with profile_api_calls(client):
    issues = client.issues.get_by_team("Engineering")
    for issue in list(issues.values())[:5]:
        comments = issue.comments  # This may hit cache

def export_team_data(client, team_name: str, format: str = "json"):
    """Export comprehensive team data for backup or analysis."""
    team_id = client.teams.get_id_by_name(team_name)
    team = client.teams.get(team_id)

    # Collect all team data
    export_data = {
        "team": team.model_dump(),
        "members": [m.model_dump() for m in client.teams.get_members(team_id)],
        "states": [s.model_dump() for s in client.teams.get_states(team_id)],
        "labels": [l.model_dump() for l in client.teams.get_labels(team_id)],
        "issues": {k: v.model_dump() for k, v in client.issues.get_by_team(team_name).items()},
        "projects": {k: v.model_dump() for k, v in client.projects.get_all(team_id=team_id).items()}
    }

    if format == "json":
        import json
        return json.dumps(export_data, indent=2, default=str)
    elif format == "csv":
        # Convert to CSV format for specific entities
        import pandas as pd
        issues_df = pd.DataFrame([issue for issue in export_data["issues"].values()])
        return issues_df.to_csv(index=False)

# Export team data
team_json = export_team_data(client, "Engineering", "json")
print(f"Exported {len(team_json)} characters of team data")