Skip to content

FSx

Utilities for working with Amazon FSx.

calculate_size_required 

calculate_size_required(bytes_required)

Calculate size of file system for the given bytes specified.

FSx file systems are created with a size of - 1.2 TB, - 2.4 TB - any multiple 2.4 TB.

Parameters:

Name Type Description Default
bytes_required int

Bytes required.

 required

Returns:

Type Description
int

Size required.
Source code in src/aibs_informatics_aws_utils/fsx.py 
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def calculate_size_required(bytes_required: int) -> int:
    """Calculate size of file system for the given bytes specified.

    FSx file systems are created with a size of
        - 1.2 TB,
        - 2.4 TB
        - any multiple 2.4 TB.

    Args:
        bytes_required (int): Bytes required.

    Returns:
        Size required.
    """
    BYTES_IN_TB = 1024 * 1024 * 1024 * 1024
    if bytes_required <= 1.2 * BYTES_IN_TB:
        return ceil(1.2 * BYTES_IN_TB)
    if bytes_required <= 2.4 * BYTES_IN_TB:
        return ceil(2.4 * BYTES_IN_TB)
    return ceil((bytes_required // (2.4 * BYTES_IN_TB) + 1) * 2.4 * BYTES_IN_TB)

get_file_system 

get_file_system(name_or_id=None, tags=None)

Get FSx file system.

Parameters:

Name Type Description Default
name_or_id Optional[FileSystemNameOrId]

File system name or id.

 None
tags Optional[Dict[str, str]]

File system tags.

 None

Returns:

Type Description
FileSystemTypeDef

File system description.
Source code in src/aibs_informatics_aws_utils/fsx.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def get_file_system(
    name_or_id: Optional[FileSystemNameOrId] = None,
    tags: Optional[Dict[str, str]] = None,
) -> FileSystemTypeDef:
    """Get FSx file system.

    Args:
        name_or_id (Optional[FileSystemNameOrId], optional): File system name or id.
        tags (Optional[Dict[str, str]], optional): File system tags.

    Returns:
        File system description.
    """
    if not name_or_id and not tags:
        raise ValueError("At least one of file_system_id, name or tags must be provided.")

    file_systems = list_file_systems(name_or_ids=[name_or_id] if name_or_id else None, tags=tags)
    if len(file_systems) > 1:
        raise ValueError(
            f"Multiple file systems found with name/id: {name_or_id} and tags: {tags}"
        )
    if not file_systems:
        raise ValueError(f"File system not found with name/id: {name_or_id} and tags: {tags}")
    return file_systems[0]

list_data_repository_associations 

list_data_repository_associations(
    name_or_id,
    filters=None,
    data_repository_paths=None,
    **kwargs
)

List data repository associations for a file system.

Parameters:

Name Type Description Default
name_or_id Optional[FileSystemIdOrName]

File system name or id.

 required
filters Optional[List[FilterTypeDef]]

Filters to apply.

 None
data_repository_paths Optional[List[str]]

Data repository paths to filter by.

 None
**kwargs

Additional arguments passed to the FSx client.

 {}

Returns:

Type Description
List[DataRepositoryAssociationTypeDef]

List of data repository associations.
Source code in src/aibs_informatics_aws_utils/fsx.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
def list_data_repository_associations(
    name_or_id: Optional[FileSystemIdOrName],
    filters: Optional[List[FilterTypeDef]] = None,
    # TODO: should I include data repository paths?
    data_repository_paths: Optional[List[str]] = None,
    **kwargs,
) -> List[DataRepositoryAssociationTypeDef]:
    """List data repository associations for a file system.

    Args:
        name_or_id (Optional[FileSystemIdOrName]): File system name or id.
        filters (Optional[List[FilterTypeDef]], optional): Filters to apply.
        data_repository_paths (Optional[List[str]], optional): Data repository paths to filter by.
        **kwargs: Additional arguments passed to the FSx client.

    Returns:
        List of data repository associations.
    """
    client = get_fsx_client(**kwargs)
    if name_or_id:
        file_system_id = resolve_file_system_ids(name_or_id)
        if filters:
            if id_filter := next(
                (filter for filter in filters if filter.get("Name") == "file-system-id"), None
            ):
                id_filter["Values"] = list(id_filter.get("Values", [])) + [str(file_system_id)]
            else:
                filters.append({"Name": "file-system-id", "Values": file_system_id})
        else:
            filters = [{"Name": "file-system-id", "Values": file_system_id}]
    if not filters:
        filters = []
    associations: List[DataRepositoryAssociationTypeDef] = []
    response = client.describe_data_repository_associations(Filters=filters)
    while response["Associations"]:
        new_associations = response["Associations"]
        if data_repository_paths:
            new_associations = [
                association
                for association in new_associations
                if association.get("DataRepositoryPath") in data_repository_paths
            ]
        associations.extend(new_associations)
        if not (next_token := response.get("NextToken")):
            break
        response = client.describe_data_repository_associations(
            Filters=filters, NextToken=next_token
        )
    return associations

list_file_systems 

list_file_systems(name_or_ids=None, tags=None, **kwargs)

List FSx file systems.

You can filter on id, name and tags.

Parameters:

Name Type Description Default
name_or_ids Optional[List[FileSystemNameOrId]]

Optionally filter on file system names or ids.

 None
tags Optional[Dict[str, str]]

Optionally filter on tags.

 None
**kwargs

Additional arguments passed to the FSx client.

 {}

Returns:

Type Description
List[FileSystemTypeDef]

List of file systems.
Source code in src/aibs_informatics_aws_utils/fsx.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
def list_file_systems(
    name_or_ids: Optional[List[FileSystemNameOrId]] = None,
    tags: Optional[Dict[str, str]] = None,
    **kwargs,
) -> List[FileSystemTypeDef]:
    """List FSx file systems.

    You can filter on id, name and tags.

    Args:
        name_or_ids (Optional[List[FileSystemNameOrId]], optional): Optionally filter on file
            system names or ids.
        tags (Optional[Dict[str, str]], optional): Optionally filter on tags.
        **kwargs: Additional arguments passed to the FSx client.

    Returns:
        List of file systems.
    """
    client = get_fsx_client(**kwargs)
    paginator = client.get_paginator("describe_file_systems")
    names, file_system_ids = split_name_and_ids(name_or_ids or [])
    if file_system_ids:
        response_iter = paginator.paginate(FileSystemIds=file_system_ids)
    else:
        response_iter = paginator.paginate()
    file_systems: List[FileSystemTypeDef] = []
    for response in response_iter:
        filtered_file_systems = response["FileSystems"]
        if not names and not tags:
            file_systems.extend(filtered_file_systems)
            continue
        old_filtered_file_systems = filtered_file_systems
        filtered_file_systems = []
        for fs in old_filtered_file_systems:
            fs_tags_dict = {tag["Key"]: tag["Value"] for tag in fs.get("Tags", [])}
            if names and fs_tags_dict.get("Name") not in names:
                continue
            if tags and not all(
                tag in fs_tags_dict and value == fs_tags_dict[tag] for tag, value in tags.items()
            ):
                continue
            filtered_file_systems.append(fs)
        file_systems.extend(filtered_file_systems)
    return file_systems

resolve_file_system_ids 

resolve_file_system_ids(*name_or_ids)

Resolve file system ids from file system names or ids.

Parameters:

Name Type Description Default
name_or_ids Tuple[FileSystemNameOrId]

File system names or ids.

 ()

Returns:

Type Description
List[FSxFileSystemId]

File system id.
Source code in src/aibs_informatics_aws_utils/fsx.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def resolve_file_system_ids(*name_or_ids: FileSystemNameOrId) -> List[FSxFileSystemId]:
    """Resolve file system ids from file system names or ids.

    Args:
        name_or_ids (Tuple[FileSystemNameOrId]): File system names or ids.

    Returns:
        File system id.
    """
    file_system_ids: List[FSxFileSystemId] = []
    for name_or_id in name_or_ids:
        name, file_system_id = split_name_and_id(name_or_id)
        if file_system_id:
            file_system_ids.append(file_system_id)
        else:
            file_system = get_file_system(name)
            if not file_system:
                raise ValueError(f"File system not found with name: {name}")
            file_system_ids.append(FSxFileSystemId(file_system.get("FileSystemId", "")))
    return file_system_ids

split_name_and_id 

split_name_and_id(id_or_name)

Identify file system identifier as name or id

Examples:

INP: NAME1 OUT: (NAME1, None)

INP: ID1 OUT: (None, ID1)

INP: None OUT: (None, None)

Parameters:

Name Type Description Default
id_or_name Optional[FileSystemNameOrId]

File system id or name.

 required

Returns:

Type Description
Tuple[Optional[str], Optional[FSxFileSystemId]]

Tuple[Optional[str], Optional[FSxFileSystemId]]: Tuple of name and id.
Source code in src/aibs_informatics_aws_utils/fsx.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def split_name_and_id(
    id_or_name: Optional[FileSystemNameOrId],
) -> Tuple[Optional[str], Optional[FSxFileSystemId]]:
    """Identify file system identifier as name or id

    Examples:
        INP: NAME1
        OUT: (NAME1, None)

        INP: ID1
        OUT: (None, ID1)

        INP: None
        OUT: (None, None)

    Args:
        id_or_name (Optional[FileSystemNameOrId]): File system id or name.


    Returns:
        Tuple[Optional[str], Optional[FSxFileSystemId]]: Tuple of name and id.
    """
    if not id_or_name:
        return None, None
    elif FSxFileSystemId.is_valid(id_or_name):
        return None, FSxFileSystemId(id_or_name)
    return id_or_name, None

split_name_and_ids 

split_name_and_ids(names_or_ids)

Split file system combined list of names and ids into separate lists.

Example

INP: [NAME1, ID1, NAME2, NAME3, ID2] OUT: ([NAME1, NAME2, NAME3], [ID1, ID2])

Parameters:

Name Type Description Default
names_or_ids List[FileSystemNameOrId]

List of names and/or ids.

 required

Returns:

Type Description
Tuple[List[str], List[FSxFileSystemId]]

Tuple[List[str], List[FSxFileSystemId]]: Tuple of name and id lists.
Source code in src/aibs_informatics_aws_utils/fsx.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def split_name_and_ids(
    names_or_ids: List[FileSystemNameOrId],
) -> Tuple[List[str], List[FSxFileSystemId]]:
    """Split file system combined list of names and ids into separate lists.

    Example:
        INP: [NAME1, ID1, NAME2, NAME3, ID2]
        OUT: ([NAME1, NAME2, NAME3], [ID1, ID2])

    Args:
        names_or_ids (List[FileSystemNameOrId]): List of names and/or ids.

    Returns:
        Tuple[List[str], List[FSxFileSystemId]]: Tuple of name and id lists.
    """
    if not names_or_ids:
        return [], []
    names, file_system_ids = zip(*(split_name_and_id(name_or_id) for name_or_id in names_or_ids))
    return [name for name in names if name], [
        file_system_id for file_system_id in file_system_ids if file_system_id
    ]