Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ dependencies = [
]

[project.optional-dependencies]
observability = [
"logfire>=4.32.1",
]
# Full — Maximal recipe (OceanBase + all LangChain providers) + all integrations
full = [
"pyobvector>=0.2.6",
Expand All @@ -34,16 +37,14 @@ full = [
"langchain-community>=0.3",
"langgraph>=0.2",
"mcp",
"opentelemetry-api>=1.20",
"opentelemetry-sdk>=1.20",
"seekvfs[observability]",
]
dev = [
"pytest>=7.4",
"pytest-asyncio>=0.23",
"ruff>=0.4",
"mypy>=1.8",
"opentelemetry-api>=1.20",
"opentelemetry-sdk>=1.20",
"seekvfs[observability]",
]

[project.urls]
Expand Down
71 changes: 0 additions & 71 deletions src/seekvfs/observability.py

This file was deleted.

4 changes: 2 additions & 2 deletions src/seekvfs/tools/langgraph.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
"""
from __future__ import annotations

from typing import Any, Optional
from typing import Any

from pydantic import BaseModel, create_model

Expand Down Expand Up @@ -39,7 +39,7 @@ def _build_args_schema(spec: ToolSpec) -> type[BaseModel]:
fields[field_name] = (py_type, ...)
else:
default = field_info.get("default", None)
fields[field_name] = (Optional[py_type], default)
fields[field_name] = (py_type | None, default)

return create_model(spec.name, **fields)

Expand Down
107 changes: 60 additions & 47 deletions src/seekvfs/vfs.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
"""
from __future__ import annotations

from collections.abc import Iterator
from contextlib import contextmanager
from typing import TYPE_CHECKING

from seekvfs.exceptions import InvalidRouteConfig, VFSError
Expand All @@ -23,7 +25,6 @@
RouteConfig,
SearchResult,
)
from seekvfs.observability import trace_span
from seekvfs.reranker import LinearReranker
from seekvfs.router import Router
from seekvfs.uri import SCHEME
Expand All @@ -33,6 +34,18 @@
from seekvfs.tools import ToolSpecSet


@contextmanager
def _instrument_vfs(name: str) -> Iterator[None]:
try:
import logfire
except ImportError:
yield
return

with logfire.span(name):
yield


class VFS:
"""Unified front door for SeekVFS storage protocol."""

Expand Down Expand Up @@ -77,25 +90,24 @@ def _validate_routes(routes: dict[str, RouteConfig]) -> None:

# ---------- main API ----------

@trace_span("vfs.write")
def write(self, path: str, content: bytes | str) -> None:
path = self._normalize(path)
_, route = self._router.resolve(path)
route["backend"].write(path, content)
with _instrument_vfs("vfs.write"):
path = self._normalize(path)
_, route = self._router.resolve(path)
route["backend"].write(path, content)

@trace_span("vfs.read")
def read(self, path: str, hint: str | None = None) -> FileData:
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].read(path, hint=hint)
with _instrument_vfs("vfs.read"):
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].read(path, hint=hint)

@trace_span("vfs.read_full")
def read_full(self, path: str) -> FileData:
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].read_full(path)
with _instrument_vfs("vfs.read_full"):
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].read_full(path)

@trace_span("vfs.search")
def search(
self,
query: str,
Expand All @@ -104,57 +116,58 @@ def search(
score_threshold: float | None = None,
) -> SearchResult:
"""Fan out across every route sequentially, then rerank."""
routes = self._router.all_routes()
if not routes:
return SearchResult(query=query, hits=[], searched_paths=[])

per_backend: list[SearchResult] = []
for prefix, route in routes:
out = route["backend"].search(
query,
path_pattern=path_pattern,
limit=limit,
score_threshold=score_threshold,
)
if prefix not in out.searched_paths:
out.searched_paths.append(prefix)
per_backend.append(out)
with _instrument_vfs("vfs.search"):
routes = self._router.all_routes()
if not routes:
return SearchResult(query=query, hits=[], searched_paths=[])

per_backend: list[SearchResult] = []
for prefix, route in routes:
out = route["backend"].search(
query,
path_pattern=path_pattern,
limit=limit,
score_threshold=score_threshold,
)
if prefix not in out.searched_paths:
out.searched_paths.append(prefix)
per_backend.append(out)

return self._reranker.merge(per_backend, limit=limit)
return self._reranker.merge(per_backend, limit=limit)

@trace_span("vfs.ls")
def ls(
self,
path: str,
pattern: str | None = None,
recursive: bool = False,
) -> list[FileInfo]:
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].ls(path, pattern=pattern, recursive=recursive)
with _instrument_vfs("vfs.ls"):
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].ls(path, pattern=pattern, recursive=recursive)

@trace_span("vfs.edit")
def edit(self, path: str, old: str, new: str) -> int:
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].edit(path, old, new)
with _instrument_vfs("vfs.edit"):
path = self._normalize(path)
_, route = self._router.resolve(path)
return route["backend"].edit(path, old, new)

@trace_span("vfs.grep")
def grep(
self,
pattern: str,
path_pattern: str | None = None,
) -> list[GrepMatch]:
results: list[GrepMatch] = []
for _, route in self._router.all_routes():
results.extend(route["backend"].grep(pattern, path_pattern=path_pattern))
return results
with _instrument_vfs("vfs.grep"):
results: list[GrepMatch] = []
for _, route in self._router.all_routes():
results.extend(route["backend"].grep(pattern, path_pattern=path_pattern))
return results

@trace_span("vfs.delete")
def delete(self, path: str) -> None:
path = self._normalize(path)
_, route = self._router.resolve(path)
route["backend"].delete(path)
with _instrument_vfs("vfs.delete"):
path = self._normalize(path)
_, route = self._router.resolve(path)
route["backend"].delete(path)

def read_batch(self, paths: list[str]) -> dict[str, FileData]:
by_backend: dict[int, tuple[object, list[str]]] = {}
Expand Down
52 changes: 33 additions & 19 deletions tests/unit/test_observability.py
Original file line number Diff line number Diff line change
@@ -1,36 +1,50 @@
from __future__ import annotations

import builtins

import pytest

from seekvfs.vfs import VFS
from tests.conftest import _StubBackend

pytest.importorskip("opentelemetry")

@pytest.fixture
def no_logfire(monkeypatch) -> None:
original_import = builtins.__import__

def guarded_import(name, globals=None, locals=None, fromlist=(), level=0):
if name == "logfire":
raise ModuleNotFoundError("No module named 'logfire'")
return original_import(name, globals, locals, fromlist, level)

monkeypatch.setattr(builtins, "__import__", guarded_import)


@pytest.fixture
def tracer_provider():
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import (
InMemorySpanExporter,
def span_exporter():
logfire = pytest.importorskip("logfire")
testing = pytest.importorskip("logfire.testing")

exporter = testing.TestExporter()
logfire.configure(
send_to_logfire=False,
console=False,
inspect_arguments=False,
additional_span_processors=[testing.SimpleSpanProcessor(exporter)],
)
yield exporter
logfire.shutdown()

exporter = InMemorySpanExporter()
provider = TracerProvider()
provider.add_span_processor(SimpleSpanProcessor(exporter))
trace.set_tracer_provider(provider)
from seekvfs import observability

observability._tracer = None
observability._otel_available = None
return exporter
def test_write_emits_logfire_span(span_exporter) -> None:
vfs = VFS(routes={"seekvfs://a/": {"backend": _StubBackend()}})
vfs.write("seekvfs://a/p", "hello")

names = [span["name"] for span in span_exporter.exported_spans_as_dict()]
assert "vfs.write" in names

def test_write_emits_span(tracer_provider) -> None:

def test_write_read_degrade_without_logfire(no_logfire) -> None:
vfs = VFS(routes={"seekvfs://a/": {"backend": _StubBackend()}})
vfs.write("seekvfs://a/p", "hello")
spans = tracer_provider.get_finished_spans()
names = {s.name for s in spans}
assert "vfs.write" in names
assert vfs.read("seekvfs://a/p").content == b"hello"
Loading