mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
remove project logic from MemoryUrl
This commit is contained in:
@@ -110,7 +110,7 @@ async def get_memory_context(
|
||||
logger.debug(
|
||||
f"Getting context for URI: `{uri}` depth: `{depth}` timeframe: `{timeframe}` max_results: `{max_results}`"
|
||||
)
|
||||
memory_url = MemoryUrl(f"memory://{config.project}/{uri}")
|
||||
memory_url = MemoryUrl(f"memory://{uri}")
|
||||
|
||||
# Parse timeframe
|
||||
since = parse(timeframe)
|
||||
|
||||
@@ -3,49 +3,52 @@
|
||||
from datetime import datetime
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
from pydantic import AnyUrl, Field, BaseModel
|
||||
from pydantic import BaseModel, field_validator, Field
|
||||
|
||||
from basic_memory.config import config
|
||||
from basic_memory.schemas.search import SearchItemType
|
||||
|
||||
"""Memory URL schema for knowledge addressing.
|
||||
|
||||
The memory:// URL scheme provides a unified way to address knowledge:
|
||||
|
||||
Examples:
|
||||
memory://specs/search/* # Pattern matching
|
||||
memory://specs/xyz # direct reference
|
||||
"""
|
||||
|
||||
|
||||
class MemoryUrl(AnyUrl):
|
||||
"""memory:// URL scheme for knowledge addressing."""
|
||||
|
||||
allowed_schemes = {"memory"}
|
||||
|
||||
# Query params
|
||||
params: Dict[str, Any] = Field(default_factory=dict) # For special modes like 'related'
|
||||
|
||||
class MemoryUrl(BaseModel):
|
||||
"""memory:// URL scheme for knowledge addressing.
|
||||
|
||||
Example URLs:
|
||||
memory://specs/search # Direct reference
|
||||
memory://specs/search/* # Pattern matching
|
||||
memory://related/xyz # Special lookup
|
||||
"""
|
||||
url: str
|
||||
path: str = ""
|
||||
|
||||
@field_validator("url")
|
||||
@classmethod
|
||||
def validate_url(cls, v: str) -> str:
|
||||
"""Validate the URL starts with memory://."""
|
||||
if isinstance(v, MemoryUrl):
|
||||
return v.url
|
||||
if not isinstance(v, str):
|
||||
raise ValueError(f"URL must be a string, got {type(v)}")
|
||||
if not v.startswith("memory://"):
|
||||
raise ValueError(f"Invalid memory URL: {v}. Must start with memory://")
|
||||
return v
|
||||
|
||||
def __init__(self, url: str, **data):
|
||||
if isinstance(url, MemoryUrl):
|
||||
url = url.url
|
||||
super().__init__(url=url, **data)
|
||||
self.path = url.removeprefix("memory://")
|
||||
|
||||
@classmethod
|
||||
def validate(cls, url: str) -> "MemoryUrl":
|
||||
"""Validate and construct a MemoryUrl."""
|
||||
|
||||
memory_url = cls(url)
|
||||
|
||||
# if the url host value is not the project name, assume the default project
|
||||
if memory_url.host != config.project:
|
||||
memory_url = cls(f"memory://{config.project}/{memory_url.host}{memory_url.path}")
|
||||
|
||||
return memory_url
|
||||
return cls(url=url)
|
||||
|
||||
def relative_path(self) -> str:
|
||||
"""Get the path without leading slash."""
|
||||
path = self.path
|
||||
return path[1:] if path.startswith("/") else path
|
||||
"""Get the path."""
|
||||
return self.path
|
||||
|
||||
def __str__(self) -> str:
|
||||
"""Convert back to URL string."""
|
||||
return f"memory://{self.host}{self.path}"
|
||||
return self.url
|
||||
|
||||
|
||||
class EntitySummary(BaseModel):
|
||||
@@ -101,4 +104,4 @@ class GraphContext(BaseModel):
|
||||
)
|
||||
|
||||
# Context metadata
|
||||
metadata: MemoryMetadata
|
||||
metadata: MemoryMetadata
|
||||
@@ -1,51 +1,53 @@
|
||||
"""Tests for MemoryUrl parsing."""
|
||||
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from basic_memory.config import config
|
||||
from basic_memory.schemas.memory import MemoryUrl
|
||||
|
||||
|
||||
def test_basic_permalink():
|
||||
"""Test basic permalink parsing."""
|
||||
url = MemoryUrl.validate(f"memory://{config.project}/specs/search")
|
||||
assert url.scheme == "memory"
|
||||
assert url.host == config.project
|
||||
assert url.path == "/specs/search"
|
||||
url = MemoryUrl.validate("memory://specs/search")
|
||||
assert str(url) == "memory://specs/search"
|
||||
assert url.path == "specs/search"
|
||||
|
||||
|
||||
def test_glob_pattern():
|
||||
"""Test pattern matching."""
|
||||
url = MemoryUrl.validate(f"memory://{config.project}/specs/search/*")
|
||||
assert url.host == config.project
|
||||
assert url.path == "/specs/search/*"
|
||||
url = MemoryUrl.validate("memory://specs/search/*")
|
||||
assert url.path == "specs/search/*"
|
||||
|
||||
|
||||
def test_related_prefix():
|
||||
"""Test related content prefix."""
|
||||
url = MemoryUrl.validate(f"memory://{config.project}/related/specs/search")
|
||||
assert url.host == config.project
|
||||
assert url.path == "/related/specs/search"
|
||||
url = MemoryUrl.validate("memory://related/specs/search")
|
||||
assert url.path == "related/specs/search"
|
||||
|
||||
|
||||
def test_context_prefix():
|
||||
"""Test context prefix."""
|
||||
url = MemoryUrl.validate(f"memory://{config.project}/context/current")
|
||||
assert url.host == config.project
|
||||
assert url.path == "/context/current"
|
||||
|
||||
url = MemoryUrl.validate("memory://context/current")
|
||||
assert url.path == "context/current"
|
||||
|
||||
|
||||
def test_complex_pattern():
|
||||
"""Test multiple glob patterns."""
|
||||
url = MemoryUrl.validate(f"memory://{config.project}/specs/*/search/*")
|
||||
assert url.host == config.project
|
||||
assert url.path == "/specs/*/search/*"
|
||||
url = MemoryUrl.validate("memory://specs/*/search/*")
|
||||
assert url.path == "specs/*/search/*"
|
||||
|
||||
|
||||
def test_path_with_no_host():
|
||||
"""Test getting path without leading slash."""
|
||||
def test_path_with_dashes():
|
||||
"""Test path with dashes and other chars."""
|
||||
url = MemoryUrl.validate("memory://file-sync-and-note-updates-implementation")
|
||||
assert url.path == "file-sync-and-note-updates-implementation"
|
||||
|
||||
|
||||
def test_invalid_url():
|
||||
"""Test URL must start with memory://."""
|
||||
with pytest.raises(ValueError, match="Invalid memory URL"):
|
||||
MemoryUrl.validate("http://specs/search")
|
||||
|
||||
|
||||
def test_str_representation():
|
||||
"""Test converting back to string."""
|
||||
url = MemoryUrl.validate("memory://specs/search")
|
||||
assert url.relative_path() == "specs/search"
|
||||
|
||||
assert str(url) == "memory://specs/search"
|
||||
@@ -137,7 +137,7 @@ async def test_find_connected_timeframe(context_service, test_graph, search_repo
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_context(context_service, test_graph):
|
||||
"""Test exact permalink lookup."""
|
||||
url = MemoryUrl(f"memory://{config.project}/test/root")
|
||||
url = MemoryUrl(f"memory://test/root")
|
||||
results = await context_service.build_context(url)
|
||||
matched_results = results["metadata"]["matched_results"]
|
||||
primary_results = results["primary_results"]
|
||||
@@ -155,7 +155,7 @@ async def test_build_context(context_service, test_graph):
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_context_pattern(context_service, test_graph):
|
||||
"""Test exact permalink lookup."""
|
||||
url = MemoryUrl("memory://not_used/test/connected*")
|
||||
url = MemoryUrl("memory://test/connected*")
|
||||
results = await context_service.build_context(url)
|
||||
matched_results = results["metadata"]["matched_results"]
|
||||
primary_results = results["primary_results"]
|
||||
@@ -168,7 +168,7 @@ async def test_build_context_pattern(context_service, test_graph):
|
||||
@pytest.mark.asyncio
|
||||
async def test_build_context_not_found(context_service):
|
||||
"""Test handling non-existent permalinks."""
|
||||
context = await context_service.build_context(MemoryUrl("memory://project/does/not/exist"))
|
||||
context = await context_service.build_context(MemoryUrl("memory://does/not/exist"))
|
||||
assert len(context["primary_results"]) == 0
|
||||
assert len(context["related_results"]) == 0
|
||||
|
||||
@@ -176,7 +176,7 @@ async def test_build_context_not_found(context_service):
|
||||
@pytest.mark.asyncio
|
||||
async def test_context_metadata(context_service, test_graph):
|
||||
"""Test metadata is correctly populated."""
|
||||
context = await context_service.build_context(MemoryUrl("memory://project/test/root"), depth=2)
|
||||
context = await context_service.build_context(MemoryUrl("memory://test/root"), depth=2)
|
||||
metadata = context["metadata"]
|
||||
assert metadata["uri"] == "test/root"
|
||||
assert metadata["depth"] == 2
|
||||
|
||||
Reference in New Issue
Block a user