Parse 'fv_name@version:feature' into (fv_name, version_number, feature_name). If no @version is present, version_number is None (meaning 'latest'). Examples: 'driver_stats:trips' -> ('driver_stats', None, 'trips') 'driver_stats@v2:trips' -> ('driver_stats', 2, 'trips')
(ref: str)
| 63 | |
| 64 | |
| 65 | def _parse_feature_ref(ref: str) -> Tuple[str, Optional[int], str]: |
| 66 | """Parse 'fv_name@version:feature' into (fv_name, version_number, feature_name). |
| 67 | |
| 68 | If no @version is present, version_number is None (meaning 'latest'). |
| 69 | Examples: |
| 70 | 'driver_stats:trips' -> ('driver_stats', None, 'trips') |
| 71 | 'driver_stats@v2:trips' -> ('driver_stats', 2, 'trips') |
| 72 | 'driver_stats@latest:trips' -> ('driver_stats', None, 'trips') |
| 73 | """ |
| 74 | import re |
| 75 | |
| 76 | colon_idx = ref.find(":") |
| 77 | if colon_idx < 0: |
| 78 | raise ValueError( |
| 79 | f"Invalid feature reference '{ref}'. Expected format: '<feature_view>:<feature>' " |
| 80 | f"or '<feature_view>@<version>:<feature>'" |
| 81 | ) |
| 82 | |
| 83 | fv_part = ref[:colon_idx] |
| 84 | feature_name = ref[colon_idx + 1 :] |
| 85 | |
| 86 | at_idx = fv_part.find("@") |
| 87 | if at_idx < 0: |
| 88 | return (fv_part, None, feature_name) |
| 89 | |
| 90 | fv_name = fv_part[:at_idx] |
| 91 | version_str = fv_part[at_idx + 1 :] |
| 92 | |
| 93 | if not version_str or version_str.lower() == "latest": |
| 94 | return (fv_name, None, feature_name) |
| 95 | |
| 96 | # Parse version number from formats like "v2", "V2" |
| 97 | match = re.match(r"^[vV](\d+)$", version_str) |
| 98 | if not match: |
| 99 | # Not a recognized version format — treat entire fv_part as the name |
| 100 | return (fv_part, None, feature_name) |
| 101 | |
| 102 | return (fv_name, int(match.group(1)), feature_name) |
| 103 | |
| 104 | |
| 105 | def _strip_version_from_ref(ref: str) -> str: |
no outgoing calls