Parameter Store Configuration

class ParameterStoreResource(ResourceWithBoto3Configuration):
    """
    Resource for accessing SSM Parameter Store.
    """
    
    def get_parameter(self, name: str, with_decryption: bool = True) -> str:
        """
        Get parameter value from Parameter Store.
        
        Parameters:
            name: Parameter name
            with_decryption: Decrypt SecureString parameters
            
        Returns:
            str: Parameter value
        """
    
    def get_parameters_by_path(
        self, 
        path: str,
        recursive: bool = True,
        with_decryption: bool = True
    ) -> Dict[str, str]:
        """
        Get parameters by path prefix.
        
        Parameters:
            path: Parameter path prefix
            recursive: Include nested paths
            with_decryption: Decrypt SecureString parameters
            
        Returns:
            Dict: Parameter name to value mapping
        """

class ParameterStoreTag:
    """
    Tag filter for Parameter Store queries.
    """
    key: str
    values: List[str]

class SSMResource(ResourceWithBoto3Configuration):
    """
    General SSM resource for broader SSM operations.
    """

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

from dagster import op, job, Definitions
from dagster_aws.ssm import ParameterStoreResource

@op(required_resource_keys={"params"})
def get_configuration(context):
    params = context.resources.params
    
    # Get individual parameter
    api_key = params.get_parameter("/app/prod/api-key")
    
    # Get all parameters for environment
    config = params.get_parameters_by_path("/app/prod/")
    
    return {
        "api_key": api_key,
        "database_url": config["/app/prod/database-url"],
        "cache_ttl": int(config["/app/prod/cache-ttl"])
    }

@job(
    resource_defs={
        "params": ParameterStoreResource(region_name="us-east-1")
    }
)
def configured_job():
    get_configuration()

defs = Definitions(jobs=[configured_job])