Skip to content

Demand Scaffolding

Handler for preparing demand execution scaffolding.

Demand execution scaffolding handler.

Provides Lambda handlers for preparing demand execution scaffolding, including file system setup and batch job configuration.

PrepareDemandScaffoldingHandler dataclass 

PrepareDemandScaffoldingHandler()

Bases: LambdaHandler[PrepareDemandScaffoldingRequest, PrepareDemandScaffoldingResponse]

Handler for preparing demand execution scaffolding.

Sets up the necessary infrastructure for demand executions including: - EFS volume configurations for scratch, shared, and tmp storage - Pre-execution data sync requests for input data - Post-execution data sync requests for output data - Batch job builder configuration

Example 
handler = PrepareDemandScaffoldingHandler.get_handler()
response = handler(event, context)

handle 

handle(request)

Prepare scaffolding for a demand execution.

Sets up EFS configurations, creates the execution context manager, and generates setup and cleanup configurations.

Parameters:

Name Type Description Default
request PrepareDemandScaffoldingRequest

Request containing demand execution details and file system configurations.

 required

Returns:

Type Description
PrepareDemandScaffoldingResponse

Response containing the updated demand execution and
PrepareDemandScaffoldingResponse

setup/cleanup configurations.
Source code in src/aibs_informatics_aws_lambda/handlers/demand/scaffolding.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def handle(self, request: PrepareDemandScaffoldingRequest) -> PrepareDemandScaffoldingResponse:
    """Prepare scaffolding for a demand execution.

    Sets up EFS configurations, creates the execution context manager,
    and generates setup and cleanup configurations.

    Args:
        request (PrepareDemandScaffoldingRequest): Request containing demand execution
            details and file system configurations.

    Returns:
        Response containing the updated demand execution and
        setup/cleanup configurations.
    """
    scratch_vol_configuration = construct_batch_efs_configuration(
        env_base=self.env_base,
        file_system=request.file_system_configurations.scratch.file_system,
        access_point=request.file_system_configurations.scratch.access_point
        if request.file_system_configurations.scratch.access_point
        else EFS_SCRATCH_ACCESS_POINT_NAME,
        container_path=request.file_system_configurations.scratch.container_path
        if request.file_system_configurations.scratch.container_path
        else f"/opt/efs{EFS_SCRATCH_PATH}",
        read_only=False,
    )

    shared_vol_configuration = construct_batch_efs_configuration(
        env_base=self.env_base,
        file_system=request.file_system_configurations.shared.file_system,
        access_point=request.file_system_configurations.shared.access_point
        if request.file_system_configurations.shared.access_point
        else EFS_SHARED_ACCESS_POINT_NAME,
        container_path=request.file_system_configurations.shared.container_path
        if request.file_system_configurations.shared.container_path
        else f"/opt/efs{EFS_SHARED_PATH}",
        read_only=True,
    )

    if request.file_system_configurations.tmp is not None:
        tmp_vol_configuration = construct_batch_efs_configuration(
            env_base=self.env_base,
            file_system=request.file_system_configurations.tmp.file_system,
            access_point=request.file_system_configurations.tmp.access_point
            if request.file_system_configurations.tmp.access_point
            else EFS_TMP_ACCESS_POINT_NAME,
            container_path=request.file_system_configurations.tmp.container_path
            if request.file_system_configurations.tmp.container_path
            else f"/opt/efs{EFS_TMP_PATH}",
            read_only=False,
        )
    else:
        tmp_vol_configuration = None

    context_manager = DemandExecutionContextManager(
        demand_execution=request.demand_execution,
        scratch_vol_configuration=scratch_vol_configuration,
        shared_vol_configuration=shared_vol_configuration,
        tmp_vol_configuration=tmp_vol_configuration,
        configuration=request.context_manager_configuration,
        env_base=self.env_base,
    )
    batch_job_builder = context_manager.batch_job_builder

    self.setup_file_system(context_manager)
    setup_configs = DemandExecutionSetupConfigs(
        data_sync_requests=[
            sync_request.from_dict(sync_request.to_dict())
            for sync_request in context_manager.pre_execution_data_sync_requests
        ],
        batch_create_request=CreateDefinitionAndPrepareArgsRequest(
            image=batch_job_builder.image,
            job_definition_name=batch_job_builder.job_definition_name,
            job_name=batch_job_builder.job_name,
            job_queue_name=context_manager.batch_job_queue_name,
            job_definition_tags=batch_job_builder.job_definition_tags,
            command=batch_job_builder.command,
            environment=batch_job_builder.environment,
            resource_requirements=batch_job_builder.resource_requirements,
            mount_points=batch_job_builder.mount_points,
            volumes=batch_job_builder.volumes,
            retry_strategy=build_retry_strategy(num_retries=5),
            privileged=batch_job_builder.privileged,
            job_role_arn=batch_job_builder.job_role_arn,
        ),
    )

    cleanup_configs = DemandExecutionCleanupConfigs(
        data_sync_requests=[
            sync_request.from_dict(sync_request.to_dict())
            for sync_request in context_manager.post_execution_data_sync_requests
        ],
        remove_data_paths_requests=context_manager.post_execution_remove_data_paths_requests,
    )

    return PrepareDemandScaffoldingResponse(
        demand_execution=context_manager.demand_execution,
        setup_configs=setup_configs,
        cleanup_configs=cleanup_configs,
    )

setup_file_system 

setup_file_system(context_manager)

Sets up working directory for file system

Parameters:

Name Type Description Default
context_manager DemandExecutionContextManager

context manager

 required
Source code in src/aibs_informatics_aws_lambda/handlers/demand/scaffolding.py 
156
157
158
159
160
161
162
def setup_file_system(self, context_manager: DemandExecutionContextManager):
    """Sets up working directory for file system

    Args:
        context_manager (DemandExecutionContextManager): context manager
    """
    working_path = context_manager.container_working_path  # noqa: F841

construct_batch_efs_configuration 

construct_batch_efs_configuration(
    env_base,
    container_path,
    file_system,
    access_point,
    read_only=False,
)

Construct a BatchEFSConfiguration for a volume.

Creates a mount point configuration based on the provided file system and access point parameters, resolving resources by tags if names are provided.

Parameters:

Name Type Description Default
env_base EnvBase

Environment base for resource name resolution.

 required
container_path Union[Path, str]

Path where the volume will be mounted in the container.

 required
file_system Optional[str]

File system ID or name (optional, resolved via tags).

 required
access_point Optional[str]

Access point ID or name (optional, resolved via tags).

 required
read_only bool

Whether the mount should be read-only.

 False

Returns:

Type Description
BatchEFSConfiguration

Configured BatchEFSConfiguration for use with AWS Batch.
Source code in src/aibs_informatics_aws_lambda/handlers/demand/scaffolding.py 
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def construct_batch_efs_configuration(
    env_base: EnvBase,
    container_path: Union[Path, str],
    file_system: Optional[str],
    access_point: Optional[str],
    read_only: bool = False,
) -> BatchEFSConfiguration:
    """Construct a BatchEFSConfiguration for a volume.

    Creates a mount point configuration based on the provided file system
    and access point parameters, resolving resources by tags if names
    are provided.

    Args:
        env_base (EnvBase): Environment base for resource name resolution.
        container_path (Union[Path, str]): Path where the volume will be mounted in the container.
        file_system (Optional[str]): File system ID or name (optional, resolved via tags).
        access_point (Optional[str]): Access point ID or name (optional, resolved via tags).
        read_only (bool): Whether the mount should be read-only.

    Returns:
        Configured BatchEFSConfiguration for use with AWS Batch.
    """
    mount_point_config = MountPointConfiguration.build(
        mount_point=container_path,
        access_point=access_point,
        file_system=file_system,
        access_point_tags={"env_base": env_base},
        file_system_tags={"env_base": env_base},
    )
    return BatchEFSConfiguration(mount_point_config=mount_point_config, read_only=read_only)