Spaces:
Build error
Build error
File size: 1,816 Bytes
9c7d451 107515b 9c7d451 107515b 9c7d451 107515b 9c7d451 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 | """Storage modules for Headroom SDK."""
from .base import Storage
from .jsonl import JSONLStorage
from .sqlite import SQLiteStorage
__all__ = [
"Storage",
"SQLiteStorage",
"JSONLStorage",
]
def create_storage(store_url: str) -> Storage:
"""
Create a storage instance from URL.
Supported URLs (built-in):
- sqlite:///path/to/file.db
- jsonl:///path/to/file.jsonl
Other schemes (e.g. postgres://) can be provided by packages that register
the setuptools entry point headroom.storage_backend with name=<scheme>.
Args:
store_url: Storage URL.
Returns:
Storage instance.
"""
if store_url.startswith("sqlite://"):
path = store_url.replace("sqlite://", "")
# Handle sqlite:/// (3 slashes for absolute path)
if path.startswith("/"):
path = path # Already absolute
return SQLiteStorage(path)
elif store_url.startswith("jsonl://"):
path = store_url.replace("jsonl://", "")
if path.startswith("/"):
path = path
return JSONLStorage(path)
else:
# Unknown scheme: try entry point headroom.storage_backend[name=scheme]
scheme = store_url.split("://", 1)[0].lower() if "://" in store_url else ""
if scheme:
try:
from importlib.metadata import entry_points
all_eps = entry_points(group="headroom.storage_backend")
ep = next((e for e in all_eps if e.name == scheme), None)
if ep is not None:
create_fn = ep.load()
result: Storage = create_fn(store_url)
return result
except Exception:
pass
# Default to SQLite (legacy behavior)
return SQLiteStorage(store_url)
|