|
8 | 8 | class RegionalResourceArnParser: |
9 | 9 | @staticmethod |
10 | 10 | def get_account_id(arn: str) -> Optional[str]: |
11 | | - if _is_arn(arn): |
12 | | - return str(arn).split(":")[4] |
13 | | - return None |
| 11 | + parts = _get_arn_parts(arn) |
| 12 | + return parts[4] if parts else None |
14 | 13 |
|
15 | 14 | @staticmethod |
16 | 15 | def get_region(arn: str) -> Optional[str]: |
17 | | - if _is_arn(arn): |
18 | | - return str(arn).split(":")[3] |
19 | | - return None |
| 16 | + parts = _get_arn_parts(arn) |
| 17 | + return parts[3] if parts else None |
20 | 18 |
|
21 | 19 | @staticmethod |
22 | 20 | def extract_dynamodb_table_name_from_arn(arn: str) -> Optional[str]: |
23 | | - resource_name = RegionalResourceArnParser.extract_resource_name_from_arn(arn) |
24 | | - if resource_name: |
25 | | - return resource_name.replace("table/", "") |
26 | | - return None |
| 21 | + parts = _get_arn_parts(arn) |
| 22 | + return parts[-1].replace("table/", "") if parts else None |
27 | 23 |
|
28 | 24 | @staticmethod |
29 | 25 | def extract_kinesis_stream_name_from_arn(arn: str) -> Optional[str]: |
30 | | - resource_name = RegionalResourceArnParser.extract_resource_name_from_arn(arn) |
31 | | - if resource_name: |
32 | | - return resource_name.replace("stream/", "") |
33 | | - return None |
| 26 | + parts = _get_arn_parts(arn) |
| 27 | + return parts[-1].replace("stream/", "") if parts else None |
34 | 28 |
|
35 | 29 | @staticmethod |
36 | 30 | def extract_resource_name_from_arn(arn: str) -> Optional[str]: |
37 | | - # Extracts the name of the resource from an arn |
38 | | - if _is_arn(arn): |
39 | | - split = arn.split(":") |
40 | | - return split[-1] |
41 | | - return None |
42 | | - |
| 31 | + parts = _get_arn_parts(arn) |
| 32 | + return parts[-1] if parts else None |
43 | 33 |
|
44 | | -def _is_arn(arn: str) -> bool: |
45 | | - # Check if arn follows the format: |
46 | | - # arn:partition:service:region:account-id:resource-type/resource-id or |
47 | | - # arn:partition:service:region:account-id:resource-type:resource-id |
48 | | - if arn is None: |
49 | | - return False |
50 | 34 |
|
51 | | - if not str(arn).startswith("arn"): |
52 | | - return False |
53 | | - |
54 | | - arn_parts = str(arn).split(":") |
55 | | - return len(arn_parts) >= 6 and is_account_id(arn_parts[4]) |
| 35 | +def _get_arn_parts(arn: str) -> Optional[list]: |
| 36 | + if not arn or not arn.startswith("arn"): |
| 37 | + return None |
| 38 | + parts = arn.split(":") |
| 39 | + return parts if len(parts) >= 6 and is_account_id(parts[4]) else None |
0 commit comments