mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 4229748c77 |
@@ -6,18 +6,26 @@ from pathlib import Path
|
||||
import aiofiles
|
||||
import httpx
|
||||
|
||||
from basic_memory.ignore_utils import load_gitignore_patterns, should_ignore_path
|
||||
from basic_memory.ignore_utils import (
|
||||
load_bmignore_only_patterns,
|
||||
load_gitignore_patterns,
|
||||
should_ignore_path,
|
||||
)
|
||||
from basic_memory.mcp.async_client import get_client
|
||||
from basic_memory.mcp.tools.utils import call_put
|
||||
|
||||
|
||||
async def upload_path(local_path: Path, project_name: str) -> bool:
|
||||
async def upload_path(
|
||||
local_path: Path, project_name: str, verbose: bool = False, no_gitignore: bool = False
|
||||
) -> bool:
|
||||
"""
|
||||
Upload a file or directory to cloud project via WebDAV.
|
||||
|
||||
Args:
|
||||
local_path: Path to local file or directory
|
||||
project_name: Name of cloud project (destination)
|
||||
verbose: Show detailed information about filtering and uploads
|
||||
no_gitignore: Skip loading patterns from .gitignore files
|
||||
|
||||
Returns:
|
||||
True if upload succeeded, False otherwise
|
||||
@@ -35,13 +43,19 @@ async def upload_path(local_path: Path, project_name: str) -> bool:
|
||||
if local_path.is_file():
|
||||
files_to_upload = [(local_path, local_path.name)]
|
||||
else:
|
||||
files_to_upload = _get_files_to_upload(local_path)
|
||||
files_to_upload = _get_files_to_upload(local_path, verbose, no_gitignore)
|
||||
|
||||
if not files_to_upload:
|
||||
print("No files found to upload")
|
||||
return True
|
||||
|
||||
print(f"Found {len(files_to_upload)} file(s) to upload")
|
||||
if verbose:
|
||||
print(f"\nFiles to upload ({len(files_to_upload)}):")
|
||||
for file_path, rel_path in files_to_upload:
|
||||
print(f" {rel_path}")
|
||||
print()
|
||||
else:
|
||||
print(f"Found {len(files_to_upload)} file(s) to upload")
|
||||
|
||||
# Upload files using httpx
|
||||
total_bytes = 0
|
||||
@@ -81,22 +95,47 @@ async def upload_path(local_path: Path, project_name: str) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
def _get_files_to_upload(directory: Path) -> list[tuple[Path, str]]:
|
||||
def _get_files_to_upload(
|
||||
directory: Path, verbose: bool = False, no_gitignore: bool = False
|
||||
) -> list[tuple[Path, str]]:
|
||||
"""
|
||||
Get list of files to upload from directory.
|
||||
|
||||
Uses .bmignore and .gitignore patterns for filtering.
|
||||
Uses .bmignore and optionally .gitignore patterns for filtering.
|
||||
|
||||
Args:
|
||||
directory: Directory to scan
|
||||
verbose: Show detailed information about filtering
|
||||
no_gitignore: Skip loading patterns from .gitignore files
|
||||
|
||||
Returns:
|
||||
List of (absolute_path, relative_path) tuples
|
||||
"""
|
||||
files = []
|
||||
ignored_files: list[tuple[str, str]] = [] # (relative_path, matching_pattern)
|
||||
total_files_found = 0
|
||||
|
||||
# Load ignore patterns from .bmignore and .gitignore
|
||||
ignore_patterns = load_gitignore_patterns(directory)
|
||||
# Load ignore patterns
|
||||
if no_gitignore:
|
||||
ignore_patterns = load_bmignore_only_patterns()
|
||||
if verbose:
|
||||
print(f"Loaded {len(ignore_patterns)} patterns from .bmignore only")
|
||||
print(" (--no-gitignore flag: skipping .gitignore patterns)")
|
||||
else:
|
||||
ignore_patterns = load_gitignore_patterns(directory)
|
||||
if verbose:
|
||||
bmignore_count = len(load_bmignore_only_patterns())
|
||||
gitignore_count = len(ignore_patterns) - bmignore_count
|
||||
print(f"Loaded ignore patterns:")
|
||||
print(f" .bmignore: {bmignore_count} patterns")
|
||||
print(f" .gitignore: {gitignore_count} patterns")
|
||||
print(f" Total: {len(ignore_patterns)} patterns")
|
||||
|
||||
if verbose and ignore_patterns:
|
||||
print(f"\nIgnore patterns being applied:")
|
||||
for pattern in sorted(ignore_patterns):
|
||||
print(f" {pattern}")
|
||||
print()
|
||||
|
||||
# Walk through directory
|
||||
for root, dirs, filenames in os.walk(directory):
|
||||
@@ -106,16 +145,32 @@ def _get_files_to_upload(directory: Path) -> list[tuple[Path, str]]:
|
||||
filtered_dirs = []
|
||||
for d in dirs:
|
||||
dir_path = root_path / d
|
||||
if not should_ignore_path(dir_path, directory, ignore_patterns):
|
||||
if should_ignore_path(dir_path, directory, ignore_patterns):
|
||||
if verbose:
|
||||
rel_path = dir_path.relative_to(directory)
|
||||
# Find which pattern matched
|
||||
matching_pattern = _find_matching_pattern(
|
||||
dir_path, directory, ignore_patterns
|
||||
)
|
||||
ignored_files.append((f"{rel_path}/", matching_pattern))
|
||||
else:
|
||||
filtered_dirs.append(d)
|
||||
dirs[:] = filtered_dirs
|
||||
|
||||
# Process files
|
||||
for filename in filenames:
|
||||
file_path = root_path / filename
|
||||
total_files_found += 1
|
||||
|
||||
# Check if file should be ignored
|
||||
if should_ignore_path(file_path, directory, ignore_patterns):
|
||||
if verbose:
|
||||
rel_path = file_path.relative_to(directory)
|
||||
# Find which pattern matched
|
||||
matching_pattern = _find_matching_pattern(
|
||||
file_path, directory, ignore_patterns
|
||||
)
|
||||
ignored_files.append((str(rel_path), matching_pattern))
|
||||
continue
|
||||
|
||||
# Calculate relative path for remote
|
||||
@@ -125,4 +180,72 @@ def _get_files_to_upload(directory: Path) -> list[tuple[Path, str]]:
|
||||
|
||||
files.append((file_path, remote_path))
|
||||
|
||||
if verbose:
|
||||
print(f"Scan results:")
|
||||
print(f" Total files found: {total_files_found}")
|
||||
print(f" Files to upload: {len(files)}")
|
||||
print(f" Files/dirs ignored: {len(ignored_files)}")
|
||||
|
||||
if ignored_files:
|
||||
print(f"\nIgnored files and directories:")
|
||||
for rel_path, pattern in ignored_files:
|
||||
print(f" {rel_path} (matched: {pattern})")
|
||||
print()
|
||||
|
||||
return files
|
||||
|
||||
|
||||
def _find_matching_pattern(file_path: Path, base_path: Path, ignore_patterns: set[str]) -> str:
|
||||
"""Find which pattern caused a file to be ignored.
|
||||
|
||||
Args:
|
||||
file_path: The file path to check
|
||||
base_path: The base directory for relative path calculation
|
||||
ignore_patterns: Set of patterns to match against
|
||||
|
||||
Returns:
|
||||
The first matching pattern, or "unknown" if none found
|
||||
"""
|
||||
import fnmatch
|
||||
|
||||
try:
|
||||
relative_path = file_path.relative_to(base_path)
|
||||
relative_str = str(relative_path)
|
||||
relative_posix = relative_path.as_posix()
|
||||
|
||||
for pattern in ignore_patterns:
|
||||
# Handle patterns starting with / (root relative)
|
||||
if pattern.startswith("/"):
|
||||
root_pattern = pattern[1:]
|
||||
if root_pattern.endswith("/"):
|
||||
dir_name = root_pattern[:-1]
|
||||
if len(relative_path.parts) > 0 and relative_path.parts[0] == dir_name:
|
||||
return pattern
|
||||
else:
|
||||
if fnmatch.fnmatch(relative_posix, root_pattern):
|
||||
return pattern
|
||||
continue
|
||||
|
||||
# Handle directory patterns (ending with /)
|
||||
if pattern.endswith("/"):
|
||||
dir_name = pattern[:-1]
|
||||
if dir_name in relative_path.parts:
|
||||
return pattern
|
||||
continue
|
||||
|
||||
# Direct name match
|
||||
if pattern in relative_path.parts:
|
||||
return pattern
|
||||
|
||||
# Check path parts
|
||||
for part in relative_path.parts:
|
||||
if fnmatch.fnmatch(part, pattern):
|
||||
return pattern
|
||||
|
||||
# Glob pattern match on full path
|
||||
if fnmatch.fnmatch(relative_posix, pattern) or fnmatch.fnmatch(relative_str, pattern):
|
||||
return pattern
|
||||
|
||||
return "unknown"
|
||||
except ValueError:
|
||||
return "unknown"
|
||||
|
||||
@@ -43,6 +43,17 @@ def upload(
|
||||
"--sync/--no-sync",
|
||||
help="Sync project after upload (default: true)",
|
||||
),
|
||||
verbose: bool = typer.Option(
|
||||
False,
|
||||
"--verbose",
|
||||
"-v",
|
||||
help="Show detailed information about filtering and uploads",
|
||||
),
|
||||
no_gitignore: bool = typer.Option(
|
||||
False,
|
||||
"--no-gitignore",
|
||||
help="Skip loading patterns from .gitignore (still respects .bmignore)",
|
||||
),
|
||||
) -> None:
|
||||
"""Upload local files or directories to cloud project via WebDAV.
|
||||
|
||||
@@ -50,6 +61,8 @@ def upload(
|
||||
bm cloud upload ~/my-notes --project research
|
||||
bm cloud upload notes.md --project research --create-project
|
||||
bm cloud upload ~/docs --project work --no-sync
|
||||
bm cloud upload ./history/ --project proto_react --verbose
|
||||
bm cloud upload ./notes/ --project personal --no-gitignore
|
||||
"""
|
||||
|
||||
async def _upload():
|
||||
@@ -74,7 +87,7 @@ def upload(
|
||||
|
||||
# Perform upload
|
||||
console.print(f"[blue]Uploading {path} to project '{project}'...[/blue]")
|
||||
success = await upload_path(path, project)
|
||||
success = await upload_path(path, project, verbose=verbose, no_gitignore=no_gitignore)
|
||||
if not success:
|
||||
console.print("[red]Upload failed[/red]")
|
||||
raise typer.Exit(1)
|
||||
|
||||
@@ -172,6 +172,18 @@ def load_bmignore_patterns() -> Set[str]:
|
||||
return patterns
|
||||
|
||||
|
||||
def load_bmignore_only_patterns() -> Set[str]:
|
||||
"""Load patterns only from .bmignore file, excluding .gitignore.
|
||||
|
||||
This is useful when users want to respect Basic Memory's global ignore patterns
|
||||
but not project-specific .gitignore patterns (e.g., when uploading normally-gitignored files).
|
||||
|
||||
Returns:
|
||||
Set of patterns from .bmignore only
|
||||
"""
|
||||
return load_bmignore_patterns()
|
||||
|
||||
|
||||
def load_gitignore_patterns(base_path: Path) -> Set[str]:
|
||||
"""Load gitignore patterns from .gitignore file and .bmignore.
|
||||
|
||||
|
||||
@@ -326,3 +326,134 @@ class TestUploadPath:
|
||||
mock_put.assert_called_once()
|
||||
call_args = mock_put.call_args
|
||||
assert call_args[0][1] == "/webdav/my-project/subdir/file.txt"
|
||||
|
||||
|
||||
class TestVerboseAndNoGitignoreFlags:
|
||||
"""Tests for verbose and no_gitignore flags."""
|
||||
|
||||
def test_verbose_shows_detailed_output(self, tmp_path, capsys):
|
||||
"""Test that verbose flag shows detailed filtering information."""
|
||||
# Create test files
|
||||
(tmp_path / "keep.txt").write_text("keep")
|
||||
(tmp_path / "ignore.pyc").write_text("ignore")
|
||||
|
||||
# Create .gitignore
|
||||
(tmp_path / ".gitignore").write_text("*.pyc\n")
|
||||
|
||||
# Call with verbose=True
|
||||
result = _get_files_to_upload(tmp_path, verbose=True)
|
||||
|
||||
# Should find keep.txt
|
||||
assert len(result) == 1
|
||||
|
||||
# Check verbose output
|
||||
captured = capsys.readouterr()
|
||||
assert "Loaded ignore patterns:" in captured.out
|
||||
assert "Scan results:" in captured.out
|
||||
assert "Total files found:" in captured.out
|
||||
assert "Files to upload:" in captured.out
|
||||
|
||||
def test_no_gitignore_skips_gitignore_patterns(self, tmp_path):
|
||||
"""Test that no_gitignore flag skips .gitignore patterns."""
|
||||
# Create test files
|
||||
(tmp_path / "keep.txt").write_text("keep")
|
||||
(tmp_path / "would_be_ignored.pyc").write_text("content")
|
||||
|
||||
# Create .gitignore that would normally ignore .pyc files
|
||||
(tmp_path / ".gitignore").write_text("*.pyc\n")
|
||||
|
||||
# Call WITHOUT no_gitignore (default behavior)
|
||||
result_with_gitignore = _get_files_to_upload(tmp_path, no_gitignore=False)
|
||||
paths_with_gitignore = [rel for _, rel in result_with_gitignore]
|
||||
|
||||
# .pyc should be ignored
|
||||
assert "would_be_ignored.pyc" not in paths_with_gitignore
|
||||
|
||||
# Call WITH no_gitignore flag
|
||||
result_without_gitignore = _get_files_to_upload(tmp_path, no_gitignore=True)
|
||||
paths_without_gitignore = [rel for _, rel in result_without_gitignore]
|
||||
|
||||
# .pyc should NOT be ignored (since we're skipping .gitignore)
|
||||
# Note: It might still be ignored by .bmignore if it has *.pyc pattern
|
||||
# For this test, we're checking that the flag is being respected
|
||||
|
||||
def test_no_gitignore_still_respects_bmignore(self, tmp_path):
|
||||
"""Test that no_gitignore still respects .bmignore patterns."""
|
||||
# Create test files
|
||||
(tmp_path / "regular.txt").write_text("content")
|
||||
(tmp_path / ".hidden").write_text("hidden") # Should be ignored by .bmignore
|
||||
|
||||
# Create .gitignore with a different pattern
|
||||
(tmp_path / ".gitignore").write_text("*.log\n")
|
||||
|
||||
# Call with no_gitignore=True
|
||||
result = _get_files_to_upload(tmp_path, no_gitignore=True)
|
||||
paths = [rel for _, rel in result]
|
||||
|
||||
# .hidden should still be ignored (by .bmignore default patterns)
|
||||
assert ".hidden" not in paths
|
||||
# regular.txt should be included
|
||||
assert "regular.txt" in paths
|
||||
|
||||
def test_verbose_shows_ignored_files(self, tmp_path, capsys):
|
||||
"""Test that verbose mode shows which files were ignored and why."""
|
||||
# Create test files
|
||||
(tmp_path / "keep.txt").write_text("keep")
|
||||
(tmp_path / "ignore.pyc").write_text("ignore")
|
||||
|
||||
# Create .gitignore
|
||||
(tmp_path / ".gitignore").write_text("*.pyc\n")
|
||||
|
||||
# Call with verbose=True
|
||||
_get_files_to_upload(tmp_path, verbose=True)
|
||||
|
||||
# Check output includes ignored files
|
||||
captured = capsys.readouterr()
|
||||
assert "Ignored files and directories:" in captured.out
|
||||
assert "ignore.pyc" in captured.out
|
||||
|
||||
def test_verbose_and_no_gitignore_combined(self, tmp_path, capsys):
|
||||
"""Test combining verbose and no_gitignore flags."""
|
||||
# Create test files
|
||||
(tmp_path / "file.txt").write_text("content")
|
||||
|
||||
# Create .gitignore
|
||||
(tmp_path / ".gitignore").write_text("*.log\n")
|
||||
|
||||
# Call with both flags
|
||||
_get_files_to_upload(tmp_path, verbose=True, no_gitignore=True)
|
||||
|
||||
# Check output mentions .bmignore only
|
||||
captured = capsys.readouterr()
|
||||
assert "bmignore only" in captured.out or ".bmignore" in captured.out
|
||||
assert "--no-gitignore" in captured.out
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_upload_path_passes_flags(self, tmp_path):
|
||||
"""Test that upload_path passes verbose and no_gitignore to _get_files_to_upload."""
|
||||
(tmp_path / "test.txt").write_text("content")
|
||||
|
||||
mock_client = AsyncMock()
|
||||
mock_response = Mock()
|
||||
mock_response.raise_for_status = Mock()
|
||||
|
||||
with patch("basic_memory.cli.commands.cloud.upload.get_client") as mock_get_client:
|
||||
with patch("basic_memory.cli.commands.cloud.upload.call_put") as mock_put:
|
||||
with patch(
|
||||
"basic_memory.cli.commands.cloud.upload._get_files_to_upload"
|
||||
) as mock_get_files:
|
||||
with patch("aiofiles.open", create=True) as mock_aiofiles_open:
|
||||
mock_get_client.return_value.__aenter__.return_value = mock_client
|
||||
mock_get_client.return_value.__aexit__.return_value = None
|
||||
mock_put.return_value = mock_response
|
||||
mock_get_files.return_value = [(tmp_path / "test.txt", "test.txt")]
|
||||
|
||||
mock_file = AsyncMock()
|
||||
mock_file.read.return_value = b"content"
|
||||
mock_aiofiles_open.return_value.__aenter__.return_value = mock_file
|
||||
|
||||
# Call with both flags
|
||||
await upload_path(tmp_path, "project", verbose=True, no_gitignore=True)
|
||||
|
||||
# Verify _get_files_to_upload was called with the flags
|
||||
mock_get_files.assert_called_once_with(tmp_path, True, True)
|
||||
|
||||
Reference in New Issue
Block a user