"""Dagster resources for AWS services."""

from typing import Optional

import dagster as dg
from dagster import ConfigurableResource, EnvVar


class AWSGlueResource(ConfigurableResource):
    """
    Dagster resource for AWS Glue.
    
    Provides a configured boto3 Glue client with environment variable support.
    
    Example:
        >>> @asset
        ... def my_asset(context, aws_glue: AWSGlueResource):
        ...     client = aws_glue.get_client()
        ...     client.start_job_run(JobName="my_job")
    """
    
    region_name: str = "us-east-2"
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    
    def get_client(self):
        """Get a configured Glue client."""
        from src.aws_clients import get_glue_client
        
        return get_glue_client(
            region_name=self.region_name,
            aws_access_key_id=self.aws_access_key_id,
            aws_secret_access_key=self.aws_secret_access_key,
        )


class AWSS3Resource(ConfigurableResource):
    """
    Dagster resource for AWS S3.
    
    Example:
        >>> @asset
        ... def my_asset(context, aws_s3: AWSS3Resource):
        ...     client = aws_s3.get_client()
        ...     client.list_buckets()
    """
    
    region_name: str = "us-east-2"
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    
    def get_client(self):
        """Get a configured S3 client."""
        from src.aws_clients import get_s3_client
        
        return get_s3_client(
            region_name=self.region_name,
            aws_access_key_id=self.aws_access_key_id,
            aws_secret_access_key=self.aws_secret_access_key,
        )


class AWSLambdaResource(ConfigurableResource):
    """
    Dagster resource for AWS Lambda.
    
    Example:
        >>> @asset
        ... def my_asset(context, aws_lambda: AWSLambdaResource):
        ...     client = aws_lambda.get_client()
        ...     client.invoke(FunctionName="my_function")
    """
    
    region_name: str = "us-east-2"
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    
    def get_client(self):
        """Get a configured Lambda client."""
        from src.aws_clients import get_lambda_client
        
        return get_lambda_client(
            region_name=self.region_name,
            aws_access_key_id=self.aws_access_key_id,
            aws_secret_access_key=self.aws_secret_access_key,
        )


# Pre-configured resources using environment variables
# Similar to how snowflake resource is configured
aws_glue = AWSGlueResource(
    region_name=EnvVar("AWS_REGION").get_value() if EnvVar("AWS_REGION").get_value() else "us-east-2",
    aws_access_key_id=EnvVar("AWS_ACCESS_KEY_ID").get_value() if EnvVar("AWS_ACCESS_KEY_ID").get_value() else None,
    aws_secret_access_key=EnvVar("AWS_SECRET_ACCESS_KEY").get_value() if EnvVar("AWS_SECRET_ACCESS_KEY").get_value() else None,
)

aws_s3 = AWSS3Resource(
    region_name=EnvVar("AWS_REGION").get_value() if EnvVar("AWS_REGION").get_value() else "us-east-2",
    aws_access_key_id=EnvVar("AWS_ACCESS_KEY_ID").get_value() if EnvVar("AWS_ACCESS_KEY_ID").get_value() else None,
    aws_secret_access_key=EnvVar("AWS_SECRET_ACCESS_KEY").get_value() if EnvVar("AWS_SECRET_ACCESS_KEY").get_value() else None,
)

aws_lambda = AWSLambdaResource(
    region_name=EnvVar("AWS_REGION").get_value() if EnvVar("AWS_REGION").get_value() else "us-east-2",
    aws_access_key_id=EnvVar("AWS_ACCESS_KEY_ID").get_value() if EnvVar("AWS_ACCESS_KEY_ID").get_value() else None,
    aws_secret_access_key=EnvVar("AWS_SECRET_ACCESS_KEY").get_value() if EnvVar("AWS_SECRET_ACCESS_KEY").get_value() else None,
)


__all__ = [
    "AWSGlueResource",
    "AWSS3Resource", 
    "AWSLambdaResource",
    "aws_glue",
    "aws_s3",
    "aws_lambda",
]