feat: path handling has to be absolute by gneeraj2001
This commit is contained in:
parent
2da35deae3
commit
d8326a7e3a
5 changed files with 14 additions and 91 deletions
|
|
@ -15,12 +15,8 @@ class BaseConfig(BaseSettings):
|
|||
@pydantic.model_validator(mode="after")
|
||||
def validate_paths(self):
|
||||
# Require absolute paths for root directories
|
||||
self.data_root_directory = ensure_absolute_path(
|
||||
self.data_root_directory, allow_relative=False
|
||||
)
|
||||
self.system_root_directory = ensure_absolute_path(
|
||||
self.system_root_directory, allow_relative=False
|
||||
)
|
||||
self.data_root_directory = ensure_absolute_path(self.data_root_directory)
|
||||
self.system_root_directory = ensure_absolute_path(self.system_root_directory)
|
||||
return self
|
||||
|
||||
langfuse_public_key: Optional[str] = os.getenv("LANGFUSE_PUBLIC_KEY")
|
||||
|
|
|
|||
|
|
@ -60,11 +60,9 @@ class GraphConfig(BaseSettings):
|
|||
|
||||
# Handle graph file path
|
||||
if values.graph_file_path:
|
||||
# Convert relative paths to absolute using system_root_directory as base
|
||||
# Check if absolute path is provided
|
||||
values.graph_file_path = ensure_absolute_path(
|
||||
values.graph_file_path,
|
||||
base_path=base_config.system_root_directory,
|
||||
allow_relative=True
|
||||
os.path.join(values.graph_file_path, values.graph_filename)
|
||||
)
|
||||
else:
|
||||
# Default path
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import os
|
||||
import pydantic
|
||||
from pathlib import Path
|
||||
from functools import lru_cache
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
|
@ -32,12 +33,11 @@ class VectorConfig(BaseSettings):
|
|||
def validate_paths(cls, values):
|
||||
base_config = get_base_config()
|
||||
|
||||
if values.vector_db_url:
|
||||
# Convert relative paths to absolute using system_root_directory as base
|
||||
# If vector_db_url is provided and is not a path skip checking if path is absolute (as it can also be a url)
|
||||
if values.vector_db_url and Path(values.vector_db_url).exists():
|
||||
# Relative path to absolute
|
||||
values.vector_db_url = ensure_absolute_path(
|
||||
values.vector_db_url,
|
||||
base_path=base_config.system_root_directory,
|
||||
allow_relative=True,
|
||||
)
|
||||
else:
|
||||
# Default path
|
||||
|
|
|
|||
|
|
@ -9,22 +9,14 @@ def get_absolute_path(path_from_root: str) -> str:
|
|||
return str(absolute_path.resolve())
|
||||
|
||||
|
||||
def ensure_absolute_path(
|
||||
path: str, base_path: Optional[str] = None, allow_relative: bool = False
|
||||
) -> str:
|
||||
"""Ensures a path is absolute, optionally converting relative paths.
|
||||
def ensure_absolute_path(path: str) -> str:
|
||||
"""Ensures a path is absolute.
|
||||
|
||||
Args:
|
||||
path: The path to validate/convert.
|
||||
base_path: Required base when converting relative paths (e.g., SYSTEM_ROOT_DIRECTORY).
|
||||
allow_relative: If False, raises error for relative paths instead of converting.
|
||||
path: The path to validate.
|
||||
|
||||
Returns:
|
||||
Absolute path as string
|
||||
|
||||
Raises:
|
||||
ValueError: If path is None; or path is relative and allow_relative is False;
|
||||
or base_path is missing/non-absolute when converting.
|
||||
"""
|
||||
if path is None:
|
||||
raise ValueError("Path cannot be None")
|
||||
|
|
@ -32,12 +24,4 @@ def ensure_absolute_path(
|
|||
if path_obj.is_absolute():
|
||||
return str(path_obj.resolve())
|
||||
|
||||
if not allow_relative:
|
||||
raise ValueError(f"Path must be absolute. Got relative path: {path}")
|
||||
|
||||
if base_path is None:
|
||||
raise ValueError("base_path must be provided when converting relative paths")
|
||||
base = Path(base_path).expanduser()
|
||||
if not base.is_absolute():
|
||||
raise ValueError("base_path must be absolute when converting relative paths")
|
||||
return str((base / path_obj).resolve())
|
||||
raise ValueError(f"Path must be absolute. Got relative path: {path}")
|
||||
|
|
|
|||
|
|
@ -1,19 +1,16 @@
|
|||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from cognee.root_dir import ensure_absolute_path
|
||||
|
||||
# …rest of your test cases using ensure_absolute_path…
|
||||
|
||||
def test_root_dir_absolute_paths():
|
||||
"""Test absolute path handling in root_dir.py"""
|
||||
# Test with absolute path
|
||||
abs_path = "C:/absolute/path" if os.name == 'nt' else "/absolute/path"
|
||||
abs_path = "C:/absolute/path" if os.name == "nt" else "/absolute/path"
|
||||
result = ensure_absolute_path(abs_path, allow_relative=False)
|
||||
assert result == str(Path(abs_path).resolve())
|
||||
|
||||
|
||||
# Test with relative path (should fail)
|
||||
rel_path = "relative/path"
|
||||
with pytest.raises(ValueError, match="must be absolute"):
|
||||
|
|
@ -22,55 +19,3 @@ def test_root_dir_absolute_paths():
|
|||
# Test with None path
|
||||
with pytest.raises(ValueError, match="cannot be None"):
|
||||
ensure_absolute_path(None)
|
||||
def test_database_relative_paths():
|
||||
"""Test relative path handling for vector and graph databases"""
|
||||
system_root = "C:/system/root" if os.name == 'nt' else "/system/root"
|
||||
|
||||
# Test with absolute path
|
||||
abs_path = "C:/data/vector.db" if os.name == 'nt' else "/data/vector.db"
|
||||
result = ensure_absolute_path(abs_path, base_path=system_root, allow_relative=True)
|
||||
assert result == str(Path(abs_path).resolve())
|
||||
|
||||
# Test with relative path (should convert to absolute)
|
||||
rel_path = "data/vector.db"
|
||||
result = ensure_absolute_path(rel_path, base_path=system_root, allow_relative=True)
|
||||
expected = str((Path(system_root) / rel_path).resolve())
|
||||
assert result == expected
|
||||
|
||||
# Test with relative base_path (should fail)
|
||||
with pytest.raises(ValueError, match="base_path must be absolute"):
|
||||
ensure_absolute_path(rel_path, base_path="relative/base", allow_relative=True)
|
||||
|
||||
# Test without base_path for relative path
|
||||
with pytest.raises(ValueError, match="base_path must be provided"):
|
||||
ensure_absolute_path(rel_path, allow_relative=True)
|
||||
def test_path_consistency():
|
||||
"""Test that paths are handled consistently across configurations"""
|
||||
system_root = "C:/system/root" if os.name == 'nt' else "/system/root"
|
||||
|
||||
# Root directories must be absolute
|
||||
data_root = "C:/data/root" if os.name == 'nt' else "/data/root"
|
||||
assert ensure_absolute_path(data_root, allow_relative=False) == str(Path(data_root).resolve())
|
||||
|
||||
# Database paths can be relative but must resolve against system_root
|
||||
db_paths = [
|
||||
# Vector DB paths
|
||||
"vector.db", # Simple relative
|
||||
"data/vector.db", # Nested relative
|
||||
"../vector.db", # Parent relative
|
||||
"./vector.db", # Current dir relative
|
||||
# Graph DB paths
|
||||
"graph.db", # Simple relative
|
||||
"data/graph/db", # Nested relative
|
||||
"../graph.db", # Parent relative
|
||||
"./graph.db", # Current dir relative
|
||||
# With different extensions
|
||||
"data/vector.lancedb", # Vector DB with extension
|
||||
"data/graph/kuzu", # Graph DB with extension
|
||||
]
|
||||
|
||||
for rel_path in db_paths:
|
||||
result = ensure_absolute_path(rel_path, base_path=system_root, allow_relative=True)
|
||||
expected = str((Path(system_root) / rel_path).resolve())
|
||||
assert result == expected, f"Failed to resolve {rel_path} correctly"
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue