feat(paths): abstract runtime path logic for frozen distribution packages (#969)

* feat(core): abstract runtime path logic for frozen distribution packages

* Address review feedback: revert browser MCP check, persistent data dir default when frozen, and add path tests
This commit is contained in:
Kfir Sadeh
2026-06-15 19:44:10 +03:00
committed by GitHub
parent 270b8570fc
commit fc3a5e555e
10 changed files with 119 additions and 9 deletions
+22 -2
View File
@@ -2,12 +2,15 @@ import os
import logging
import sqlite3
from datetime import datetime, timezone
from pathlib import Path
from sqlalchemy import event, create_engine, Column, String, Text, Boolean, DateTime, Integer, ForeignKey, JSON, Index, func, text
from sqlalchemy.engine import Engine
from sqlalchemy.types import TypeDecorator
from sqlalchemy.ext.declarative import declarative_base, declared_attr
from sqlalchemy.orm import relationship, sessionmaker, backref
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
# Create base class for declarative models
@@ -29,9 +32,26 @@ class TimestampMixin:
def updated_at(cls):
return Column(DateTime, default=utcnow_naive, onupdate=utcnow_naive, nullable=False)
# Get database URL from environment, default to SQLite in DATA_DIR
# Ensure the writable data directory exists before SQLite connects.
from src.constants import DATA_DIR, AUTH_FILE, MEMORY_FILE, USER_PREFS_FILE, SETTINGS_FILE
DATABASE_URL = os.getenv("DATABASE_URL", f"sqlite:///{DATA_DIR}/app.db")
Path(DATA_DIR).mkdir(parents=True, exist_ok=True)
def _default_database_url() -> str:
return f"sqlite:///{Path(DATA_DIR) / 'app.db'}"
def _normalize_sqlite_url(url: str) -> str:
if not url.startswith("sqlite:///"):
return url
db_path = url.replace("sqlite:///", "", 1)
if db_path == ":memory:" or os.path.isabs(db_path):
return url
return f"sqlite:///{(Path(get_app_root()) / db_path).resolve().as_posix()}"
# Get database URL from environment, default to SQLite in DATA_DIR
DATABASE_URL = _normalize_sqlite_url(os.getenv("DATABASE_URL", _default_database_url()))
# Create engine
engine = create_engine(
+1
View File
@@ -9,6 +9,7 @@ from pathlib import Path
from fastapi import APIRouter, HTTPException, Form, Depends
from core.constants import EMBEDDING_ENDPOINT_FILE, FASTEMBED_CACHE_DIR
from core.middleware import require_admin
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
+3 -2
View File
@@ -14,6 +14,7 @@ import subprocess
import sys
from core.platform_compat import IS_WINDOWS, which_tool
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
@@ -81,7 +82,7 @@ _BUILTIN_NPX_SERVERS = {
"name": "Built-in: Browser",
"command": "npx",
"args": ["-y", "@playwright/mcp@latest", "--headless", "--caps", "vision"],
},
}
}
# Global flag to disable MCP if there are compatibility issues
@@ -94,7 +95,7 @@ async def register_builtin_servers(mcp_manager):
logger.info("Built-in MCP servers disabled via ODYSSEUS_DISABLE_MCP")
return
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
base_dir = get_app_root()
python = sys.executable
async def _connect_python_server(server_id: str, script_path: str, name: str):
+3 -2
View File
@@ -5,6 +5,7 @@ from pydantic_settings import BaseSettings, SettingsConfigDict
from pydantic import Field, field_validator
from src.constants import DATA_DIR as _DATA_DIR_CONST
from src.runtime_paths import get_app_root
# Cross-platform OS flag, exposed here so callers can `from src.config import
# IS_WINDOWS`. Defined locally (a trivial `os.name == "nt"`) rather than imported
@@ -19,7 +20,7 @@ IS_WINDOWS = os.name == "nt"
class DataConfig(BaseSettings):
"""Configuration for data storage and file handling."""
# Base directory
base_dir: Path = Field(default=Path(__file__).parent.parent, description="Base directory for the application")
base_dir: Path = Field(default=Path(get_app_root()), description="Base directory for the application")
# Data paths
data_dir: Path = Field(default=Path(_DATA_DIR_CONST), description="Main data directory")
@@ -138,7 +139,7 @@ class AppConfig(BaseSettings):
if isinstance(v, dict) and "base_dir" in v:
base_dir = v["base_dir"]
else:
base_dir = Path(__file__).parent.parent
base_dir = Path(get_app_root())
# Convert string paths to Path objects relative to base_dir
data_dir = Path(_DATA_DIR_CONST)
+4 -2
View File
@@ -2,12 +2,14 @@
"""Application-wide constants and configuration values."""
import os
from src.runtime_paths import get_app_root, get_default_data_dir
APP_VERSION = "1.0.0"
# Base paths
BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + "/"
BASE_DIR = os.path.join(get_app_root(), "")
STATIC_DIR = os.path.join(BASE_DIR, "static")
DATA_DIR = os.getenv("ODYSSEUS_DATA_DIR", os.path.join(BASE_DIR, "data"))
DATA_DIR = os.getenv("ODYSSEUS_DATA_DIR", get_default_data_dir())
# Data file paths
# Single source of truth: every persisted file/dir lives under DATA_DIR, which
+2
View File
@@ -31,6 +31,8 @@ import numpy as np
import httpx
from typing import List, Optional
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
_DEFAULT_MODEL = "all-minilm:l6-v2"
+3 -1
View File
@@ -11,6 +11,8 @@ import os
import re
from typing import Any, Dict, List, Optional, Set, Tuple
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
def _format_mcp_connection_error(name: str, command: str = "", args: Optional[List[str]] = None, error: Exception = None) -> str:
@@ -508,7 +510,7 @@ class McpManager:
return False
script_rel, name = _BUILTIN_SERVERS[server_id]
base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
base_dir = get_app_root()
script_path = os.path.join(base_dir, script_rel)
# Clean up old connection
+1
View File
@@ -7,6 +7,7 @@ import time
from pathlib import Path
from src.constants import RAG_DIR
from src.runtime_paths import get_app_root
logger = logging.getLogger(__name__)
+30
View File
@@ -0,0 +1,30 @@
"""Helpers for resolving runtime paths in source and frozen builds."""
import os
import sys
def get_app_root() -> str:
"""Return the app root directory.
In normal source runs, this is the repository root. In a frozen Windows
build, it is the bundle content root (PyInstaller's internal directory)
so bundled runtime folders like `static/`, `scripts/`, and `data/` stay
together with the executable payload.
"""
if getattr(sys, "frozen", False):
return getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(sys.executable)))
return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
def get_default_data_dir() -> str:
"""Return the default path to the data directory.
In normal runs, this is a 'data' subdirectory under the app root.
In frozen builds, it is a persistent user directory (~/.odysseus/data)
to prevent SQLite databases and other persistent files from being
written to the ephemeral, temporary extraction bundle directory.
"""
if getattr(sys, "frozen", False):
return os.path.join(os.path.expanduser("~"), ".odysseus", "data")
return os.path.join(get_app_root(), "data")
+50
View File
@@ -0,0 +1,50 @@
import os
import sys
from unittest import mock
import pytest
from src.runtime_paths import get_app_root, get_default_data_dir
def test_get_app_root_normal_run():
"""Verify that get_app_root returns the repository root parent of src/ when not frozen."""
with mock.patch.object(sys, "frozen", False, create=True):
app_root = get_app_root()
# Verify it is a valid directory path and matches expected parent structure
assert os.path.isdir(app_root)
assert os.path.exists(os.path.join(app_root, "src"))
def test_get_app_root_frozen_with_meipass():
"""Verify that get_app_root returns the sys._MEIPASS directory when frozen by PyInstaller."""
mock_meipass = os.path.abspath("mock_meipass_dir")
with mock.patch.object(sys, "frozen", True, create=True), \
mock.patch.object(sys, "_MEIPASS", mock_meipass, create=True):
app_root = get_app_root()
assert app_root == mock_meipass
def test_get_app_root_frozen_without_meipass():
"""Verify that get_app_root falls back to the sys.executable parent directory when frozen but _MEIPASS is absent."""
mock_exe_path = os.path.join(os.path.abspath("mock_exe_dir"), "Odysseus.exe")
with mock.patch.object(sys, "frozen", True, create=True), \
mock.patch.object(sys, "executable", mock_exe_path, create=True):
# Remove sys._MEIPASS if it exists in the test process environment
if hasattr(sys, "_MEIPASS"):
delattr(sys, "_MEIPASS")
app_root = get_app_root()
assert app_root == os.path.abspath("mock_exe_dir")
def test_get_default_data_dir_normal():
"""Verify that get_default_data_dir resolves to get_app_root() / 'data' when not frozen."""
with mock.patch.object(sys, "frozen", False, create=True):
res = get_default_data_dir()
assert res == os.path.join(get_app_root(), "data")
def test_get_default_data_dir_frozen():
"""Verify that get_default_data_dir resolves to a persistent user path under ~ when frozen."""
with mock.patch.object(sys, "frozen", True, create=True):
res = get_default_data_dir()
expected = os.path.join(os.path.expanduser("~"), ".odysseus", "data")
assert res == expected