Source code for blissoda.xrpd.compatibility

from blissdata.beacon.data import BeaconData
from packaging.version import Version
from importlib.metadata import version

if Version(version("blissdata")) >= Version("1.1"):
    from pydantic import Field  # noqa
else:
    from pydantic.v1 import Field  # noqa


[docs] def get_redis_db_url(): if Version(version("blissdata")) >= Version("1.0"): return BeaconData().get_redis_db() raw_url = BeaconData().get_redis_db() host, url = raw_url.split(":") if url.endswith("sock"): return f"unix://{url}" else: return f"redis://{url}"