Dagster AWS

# S3 functionality
from dagster_aws.s3 import S3Resource, s3_resource, S3PickleIOManager

# ECS functionality
from dagster_aws.ecs import EcsRunLauncher, ecs_executor

# EMR functionality
from dagster_aws.emr import EmrJobRunner, emr_pyspark_step_launcher

# Redshift functionality
from dagster_aws.redshift import RedshiftResource, redshift_resource

# Other services
from dagster_aws.athena import AthenaResource
from dagster_aws.cloudwatch import cloudwatch_logger
from dagster_aws.secretsmanager import SecretsManagerResource
from dagster_aws.ssm import ParameterStoreResource
from dagster_aws.pipes import PipesECSClient, PipesS3ContextInjector

from dagster import Definitions, asset
from dagster_aws.s3 import S3Resource, S3PickleIOManager
from dagster_aws.ecs import ecs_executor

# Configure S3 resource
s3_resource = S3Resource(region_name="us-west-2")

# Configure S3 I/O manager for asset storage
s3_io_manager = S3PickleIOManager(
    s3_resource=s3_resource,
    s3_bucket="my-data-bucket"
)

# Use ECS executor for distributed computation
@asset
def my_data_asset():
    return [1, 2, 3, 4, 5]

# Define deployment with AWS resources
defs = Definitions(
    assets=[my_data_asset],
    resources={
        "s3": s3_resource,
        "io_manager": s3_io_manager,
    },
    executors={
        "ecs": ecs_executor.configured({
            "cluster": "my-dagster-cluster",
            "subnets": ["subnet-12345"],
            "security_group_ids": ["sg-67890"]
        })
    }
)

class S3Resource(ResourceWithBoto3Configuration):
    def get_client(self): ...

class S3PickleIOManager(ConfigurableIOManager):
    def load_input(self, context): ...
    def handle_output(self, context, obj): ...

def s3_resource(**kwargs) -> S3Resource: ...
def s3_pickle_io_manager(**kwargs): ...

class EcsRunLauncher(RunLauncher):
    def launch_run(self, context): ...

def ecs_executor(**kwargs): ...

class EcsEventualConsistencyTimeout(Exception): ...

class EmrJobRunner:
    def run_job_flow(self, **kwargs): ...
    def wait_for_completion(self): ...

def emr_pyspark_step_launcher(**kwargs): ...

class EmrError(Exception): ...
class EmrClusterState(Enum): ...
class EmrStepState(Enum): ...

class RedshiftResource(ResourceWithBoto3Configuration):
    def get_connection(self): ...
    def execute_query(self, query: str): ...

def redshift_resource(**kwargs) -> RedshiftResource: ...

class RedshiftError(Exception): ...

class AthenaResource(ResourceWithBoto3Configuration):
    def execute_query(self, query: str): ...
    def get_query_results(self, execution_id: str): ...

def athena_resource(**kwargs) -> AthenaResource: ...

class AthenaError(Exception): ...
class AthenaTimeout(Exception): ...

def cloudwatch_logger(**kwargs): ...

class SecretsManagerResource(ResourceWithBoto3Configuration):
    def get_secret(self, secret_id: str): ...

def secretsmanager_resource(**kwargs) -> SecretsManagerResource: ...
def get_secrets_from_arns(arns: list) -> dict: ...
def get_tagged_secrets(tags: dict) -> dict: ...

class ParameterStoreResource(ResourceWithBoto3Configuration):
    def get_parameter(self, name: str): ...
    def get_parameters_by_path(self, path: str): ...

class SSMResource(ResourceWithBoto3Configuration): ...

def parameter_store_resource(**kwargs) -> ParameterStoreResource: ...
def ssm_resource(**kwargs) -> SSMResource: ...

class ECRPublicResource(ResourceWithBoto3Configuration):
    def get_authorization_token(self): ...

def ecr_public_resource(**kwargs) -> ECRPublicResource: ...

class RDSResource(ResourceWithBoto3Configuration):
    def get_connection(self): ...

class PipesECSClient(PipesClient):
    def run(self, context, **kwargs): ...

class PipesLambdaClient(PipesClient):
    def run(self, context, **kwargs): ...

class PipesS3ContextInjector(PipesContextInjector):
    def inject_context(self, context): ...

class PipesCloudWatchLogReader(PipesLogReader):
    def read_logs(self): ...

class ResourceWithBoto3Configuration(ConfigurableResource):
    """
    Base resource class for AWS services using boto3 with standard configuration options.
    """
    region_name: Optional[str] = None
    max_attempts: int = 5
    profile_name: Optional[str] = None
    use_ssl: bool = True
    endpoint_url: Optional[str] = None
    verify: Optional[bool] = True
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    aws_session_token: Optional[str] = None

class ResourceWithS3Configuration(ConfigurableResource):
    """
    Base resource class for S3-specific services with S3-focused configuration options.
    """
    use_unsigned_session: bool = False
    region_name: Optional[str] = None
    endpoint_url: Optional[str] = None
    max_attempts: int = 5
    profile_name: Optional[str] = None
    use_ssl: bool = True
    verify: Optional[bool] = None
    aws_access_key_id: Optional[str] = None
    aws_secret_access_key: Optional[str] = None
    aws_session_token: Optional[str] = None

def construct_boto_client_retry_config(max_attempts: int) -> dict:
    """
    Construct retry configuration for boto3 clients.
    
    Parameters:
        max_attempts: Maximum number of retry attempts
        
    Returns:
        dict: Boto3 retry configuration
    """