def has_heartbeat_capable_version_tag()

in services/utils/__init__.py [0:0]

• 12 lines of code
• 9 McCabe index (conditional complexity)

def has_heartbeat_capable_version_tag(system_tags):
    """Check client version tag whether it is known to support heartbeats or not"""
    try:
        # only parse for the major.minor.patch version and disregard any trailing bits that might cause issues with comparison.
        version_tags = [tag for tag in system_tags if tag.startswith('metaflow_version:')]
        if not version_tags:
            return False
        # match versions: major | major.minor | major.minor.patch
        ver_string = re.match(r"(0|\d+)(\.(0|\d+))*", version_tags[0].lstrip("metaflow_version:"))[0]
        version = parse(ver_string)

        if version >= Version("1") and version < Version("2"):
            return version >= Version("1.14.0")

        return version >= Version("2.2.12")
    except Exception:
        # Treat non-standard versions as heartbeat-enabled by default
        return True