"""AWS client configuration with environment variable support.""" import os from typing import Optional, Dict, Any import boto3 from botocore.config import Config from botocore.exceptions import ClientError, NoCredentialsError def get_aws_config( region_name: Optional[str] = None, max_attempts: int = 3, mode: str = "adaptive", ) -> Config: """ Get AWS client configuration. Args: region_name: AWS region (defaults to env var or us-east-2) max_attempts: Maximum retry attempts mode: Retry mode (standard, legacy, adaptive) Returns: boto3 Config object """ return Config( region_name=region_name or os.environ.get("AWS_REGION", "us-east-2"), retries={"max_attempts": max_attempts, "mode": mode}, ) def get_glue_client( region_name: Optional[str] = None, aws_access_key_id: Optional[str] = None, aws_secret_access_key: Optional[str] = None, ): """ Create and return an AWS Glue client with environment variable support. Args: region_name: AWS region (defaults to AWS_REGION env var or us-east-2) aws_access_key_id: AWS access key (defaults to AWS_ACCESS_KEY_ID env var) aws_secret_access_key: AWS secret key (defaults to AWS_SECRET_ACCESS_KEY env var) Returns: boto3 Glue client Environment Variables: - AWS_REGION: AWS region (default: us-east-2) - AWS_ACCESS_KEY_ID: AWS access key - AWS_SECRET_ACCESS_KEY: AWS secret access key Example: >>> client = get_glue_client() >>> # Or with explicit credentials >>> client = get_glue_client(region_name="us-west-2") """ config = get_aws_config(region_name=region_name) client_params = {"config": config} # Use provided credentials or fall back to environment variables # If neither is provided, boto3 will use default credential chain # (env vars, credentials file, IAM role, etc.) if aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID"): client_params["aws_access_key_id"] = aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID") if aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY"): client_params["aws_secret_access_key"] = aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY") return boto3.client("glue", **client_params) def get_s3_client( region_name: Optional[str] = None, aws_access_key_id: Optional[str] = None, aws_secret_access_key: Optional[str] = None, ): """ Create and return an AWS S3 client with environment variable support. Args: region_name: AWS region (defaults to AWS_REGION env var or us-east-2) aws_access_key_id: AWS access key (defaults to AWS_ACCESS_KEY_ID env var) aws_secret_access_key: AWS secret key (defaults to AWS_SECRET_ACCESS_KEY env var) Returns: boto3 S3 client """ config = get_aws_config(region_name=region_name) client_params = {"config": config} if aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID"): client_params["aws_access_key_id"] = aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID") if aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY"): client_params["aws_secret_access_key"] = aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY") return boto3.client("s3", **client_params) def get_lambda_client( region_name: Optional[str] = None, aws_access_key_id: Optional[str] = None, aws_secret_access_key: Optional[str] = None, ): """ Create and return an AWS Lambda client with environment variable support. Args: region_name: AWS region (defaults to AWS_REGION env var or us-east-2) aws_access_key_id: AWS access key (defaults to AWS_ACCESS_KEY_ID env var) aws_secret_access_key: AWS secret key (defaults to AWS_SECRET_ACCESS_KEY env var) Returns: boto3 Lambda client """ config = get_aws_config(region_name=region_name) client_params = {"config": config} if aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID"): client_params["aws_access_key_id"] = aws_access_key_id or os.environ.get("AWS_ACCESS_KEY_ID") if aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY"): client_params["aws_secret_access_key"] = aws_secret_access_key or os.environ.get("AWS_SECRET_ACCESS_KEY") return boto3.client("lambda", **client_params) """ Test AWS Glue client connection. Args: client: Optional Glue client to test (defaults to creating a new one) Returns: Dictionary with connection status and details Example: >>> result = test_glue_connection() >>> if result["connected"]: >>> print(f"Connected to Glue in {result['region']}") """ if client is None: client = get_glue_client() try: # Try to list jobs (with limit 1 to minimize API calls) response = client.list_jobs(MaxResults=1) return { "connected": True, "service": "glue", "region": client.meta.region_name, "message": "Successfully connected to AWS Glue", "jobs_found": len(response.get("JobNames", [])), } except NoCredentialsError: return { "connected": False, "service": "glue", "error": "No AWS credentials found", "message": "Please set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables", } except ClientError as e: return { "connected": False, "service": "glue", "error": str(e), "error_code": e.response.get("Error", {}).get("Code"), "message": f"Failed to connect to AWS Glue: {e}", } except Exception as e: return { "connected": False, "service": "glue", "error": str(e), "message": f"Unexpected error connecting to AWS Glue: {e}", } """ Test AWS S3 client connection. Args: client: Optional S3 client to test (defaults to creating a new one) Returns: Dictionary with connection status and details Example: >>> result = test_s3_connection() >>> if result["connected"]: >>> print(f"Connected to S3 in {result['region']}") """ if client is None: client = get_s3_client() try: # Try to list buckets response = client.list_buckets() return { "connected": True, "service": "s3", "region": client.meta.region_name, "message": "Successfully connected to AWS S3", "buckets_found": len(response.get("Buckets", [])), } except NoCredentialsError: return { "connected": False, "service": "s3", "error": "No AWS credentials found", "message": "Please set AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY environment variables", } except ClientError as e: return { "connected": False, "service": "s3", "error": str(e), "error_code": e.response.get("Error", {}).get("Code"), "message": f"Failed to connect to AWS S3: {e}", } except Exception as e: return { "connected": False, "service": "s3", "error": str(e), "message": f"Unexpected error connecting to AWS S3: {e}", } """ Test all AWS client connections. Returns: Dictionary with connection status for all services Example: >>> results = test_all_connections() >>> for service, result in results.items(): >>> print(f"{service}: {'✓' if result['connected'] else '✗'}") """ return { "glue": test_glue_connection(), "s3": test_s3_connection(), } # Default clients using environment variables # These maintain backward compatibility with existing code glue_client = get_glue_client(region_name="us-east-2") s3_client = get_s3_client() lambda_client = get_lambda_client() __all__ = [ "glue_client", "s3_client", "lambda_client", "get_glue_client", "get_s3_client", "get_lambda_client", "get_aws_config", ]