admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: enforce PLC0415 (import-outside-top-level) for source code

Browse Source 
Enable ruff PLC0415 rule for all source files (tests excluded via
per-file-ignores). Move 136 inline imports to top-level across 33 files.
8 imports suppressed with noqa for legitimate reasons: circular
dependencies (client/__init__.py, context.py), optional dependency
guards (app.py document processors, auth/userinfo_routes.py), and
post-env-setup imports (smithery_main.py).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-02-20 08:04:50 +01:00
co-authored by Claude Opus 4.6
parent 81efa6e263
commit a11ae9c027
32 changed files with 157 additions and 348 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/api/management.py
+6 -13
View File
@@ -20,9 +20,15 @@ import time
from importlib.metadata import version
from typing import Any
from qdrant_client.models import Filter
from starlette.requests import Request
from starlette.responses import JSONResponse
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.config_validators import AuthMode, detect_auth_mode
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
@@ -196,16 +202,12 @@ async def get_server_status(request: Request) -> JSONResponse:
# Public endpoint - no authentication required
# Get configuration
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
# Calculate uptime
uptime_seconds = int(time.time() - _server_start_time)
# Determine auth mode using proper mode detection
from nextcloud_mcp_server.config_validators import AuthMode, detect_auth_mode
mode = detect_auth_mode(settings)
# Map deployment mode to auth_mode for API response
@@ -266,8 +268,6 @@ async def get_vector_sync_status(request: Request) -> JSONResponse:
"""
# Public endpoint - no authentication required
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
if not settings.vector_sync_enabled:
return JSONResponse(
@@ -299,11 +299,6 @@ async def get_vector_sync_status(request: Request) -> JSONResponse:
# Get Qdrant client and query indexed count
indexed_count = 0
try:
from qdrant_client.models import Filter
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
qdrant_client = await get_qdrant_client()
# Count documents in collection, excluding placeholders
@@ -375,8 +370,6 @@ async def get_user_session(request: Request) -> JSONResponse:
# Check if offline access is enabled
# Use settings.enable_offline_access which handles both ENABLE_BACKGROUND_OPERATIONS (new)
# and ENABLE_OFFLINE_ACCESS (deprecated) environment variables
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
enable_offline_access = settings.enable_offline_access
nextcloud_mcp_server/api/passwords.py
+5 -13
View File
@@ -15,18 +15,16 @@ import logging
import re
import time
from collections import defaultdict
from typing import TYPE_CHECKING
import httpx
from starlette.requests import Request
from starlette.responses import JSONResponse
from ..http import nextcloud_httpx_client
if TYPE_CHECKING:
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.api.management import _sanitize_error_for_client
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.config import get_settings
from ..http import nextcloud_httpx_client
logger = logging.getLogger(__name__)
@@ -158,7 +156,7 @@ def _extract_basic_auth(
return username, password, None
async def _get_app_password_storage(request: Request) -> "RefreshTokenStorage":
async def _get_app_password_storage(request: Request) -> RefreshTokenStorage:
"""Get or initialize RefreshTokenStorage for app password operations.
Checks app.state.storage first, then falls back to creating from environment.
@@ -170,8 +168,6 @@ async def _get_app_password_storage(request: Request) -> "RefreshTokenStorage":
Returns:
Initialized RefreshTokenStorage instance
"""
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
storage = getattr(request.app.state, "storage", None)
if not storage:
@@ -202,8 +198,6 @@ async def provision_app_password(request: Request) -> JSONResponse:
- Only the user who owns the password can provision it
- Rate limited to prevent brute-force attacks
"""
from nextcloud_mcp_server.config import get_settings
# Get user_id from path
path_user_id = request.path_params.get("user_id")
if not path_user_id:
@@ -364,8 +358,6 @@ async def delete_app_password(request: Request) -> JSONResponse:
Requires BasicAuth with the user's credentials.
"""
from nextcloud_mcp_server.config import get_settings
# Get user_id from path
path_user_id = request.path_params.get("user_id")
if not path_user_id:
nextcloud_mcp_server/api/visualization.py
+13 -50
View File
@@ -11,13 +11,10 @@ All endpoints require OAuth bearer token authentication via UnifiedTokenVerifier
import base64
import logging
from typing import TYPE_CHECKING, Any
from typing import Any
import pymupdf
if TYPE_CHECKING:
pass
from qdrant_client.models import FieldCondition, Filter, MatchValue
from starlette.requests import Request
from starlette.responses import JSONResponse
@@ -29,6 +26,17 @@ from nextcloud_mcp_server.api.management import (
extract_bearer_token,
validate_token_and_get_user,
)
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.embedding.service import get_embedding_service
from nextcloud_mcp_server.search import (
BM25HybridSearchAlgorithm,
SemanticSearchAlgorithm,
)
from nextcloud_mcp_server.search.context import get_chunk_with_context
from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.visualization import compute_pca_coordinates
logger = logging.getLogger(__name__)
@@ -68,8 +76,6 @@ async def unified_search(request: Request) -> JSONResponse:
Requires OAuth bearer token for user filtering.
"""
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
if not settings.vector_sync_enabled:
return JSONResponse(
@@ -144,12 +150,6 @@ async def unified_search(request: Request) -> JSONResponse:
if fusion not in valid_fusions:
fusion = "rrf"
# Execute search using the appropriate algorithm
from nextcloud_mcp_server.search import (
BM25HybridSearchAlgorithm,
SemanticSearchAlgorithm,
)
# Select search algorithm
if algorithm == "semantic":
search_algo = SemanticSearchAlgorithm(score_threshold=score_threshold)
@@ -254,17 +254,9 @@ async def unified_search(request: Request) -> JSONResponse:
# Optional PCA coordinates
if include_pca and len(paginated_results) >= 2:
try:
from nextcloud_mcp_server.vector.visualization import (
compute_pca_coordinates,
)
if search_algo.query_embedding is not None:
query_embedding = search_algo.query_embedding
else:
from nextcloud_mcp_server.embedding.service import (
get_embedding_service,
)
embedding_service = get_embedding_service()
query_embedding = await embedding_service.embed(query)
@@ -305,8 +297,6 @@ async def vector_search(request: Request) -> JSONResponse:
Requires OAuth bearer token for user filtering.
"""
from nextcloud_mcp_server.config import get_settings
settings = get_settings()
if not settings.vector_sync_enabled:
return JSONResponse(
@@ -354,12 +344,6 @@ async def vector_search(request: Request) -> JSONResponse:
if fusion not in valid_fusions:
fusion = "rrf"
# Execute search using the appropriate algorithm
from nextcloud_mcp_server.search import (
BM25HybridSearchAlgorithm,
SemanticSearchAlgorithm,
)
# Select search algorithm
if algorithm == "semantic":
search_algo = SemanticSearchAlgorithm(score_threshold=score_threshold)
@@ -428,18 +412,10 @@ async def vector_search(request: Request) -> JSONResponse:
# Compute PCA coordinates for visualization using shared function
if include_pca and len(all_results) >= 2:
try:
from nextcloud_mcp_server.vector.visualization import (
compute_pca_coordinates,
)
# Get query embedding from search algorithm or generate it
if search_algo.query_embedding is not None:
query_embedding = search_algo.query_embedding
else:
from nextcloud_mcp_server.embedding.service import (
get_embedding_service,
)
embedding_service = get_embedding_service()
query_embedding = await embedding_service.embed(query)
@@ -549,9 +525,6 @@ async def get_chunk_context(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Initialize authenticated Nextcloud client
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.search.context import get_chunk_with_context
async with NextcloudClient.from_token(
base_url=nextcloud_host, token=token, username=user_id
) as nc_client:
@@ -581,14 +554,6 @@ async def get_chunk_context(request: Request) -> JSONResponse:
if doc_type == "file":
try:
from qdrant_client.models import FieldCondition, Filter, MatchValue
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.vector.placeholder import (
get_placeholder_filter,
)
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
settings = get_settings()
qdrant_client = await get_qdrant_client()
@@ -735,8 +700,6 @@ async def get_pdf_preview(request: Request) -> JSONResponse:
raise ValueError("Nextcloud host not configured")
# Download PDF via WebDAV using user's token
from nextcloud_mcp_server.client import NextcloudClient
async with NextcloudClient.from_token(
base_url=nextcloud_host, token=token, username=user_id
) as nc_client:
nextcloud_mcp_server/api/webhooks.py
+1 -6
View File
@@ -18,6 +18,7 @@ from nextcloud_mcp_server.api.management import (
extract_bearer_token,
validate_token_and_get_user,
)
from nextcloud_mcp_server.client.webhooks import WebhooksClient
from ..http import nextcloud_httpx_client
@@ -115,8 +116,6 @@ async def list_webhooks(request: Request) -> JSONResponse:
)
try:
from nextcloud_mcp_server.client.webhooks import WebhooksClient
# Get Bearer token from request
token = extract_bearer_token(request)
if not token:
@@ -180,8 +179,6 @@ async def create_webhook(request: Request) -> JSONResponse:
)
try:
from nextcloud_mcp_server.client.webhooks import WebhooksClient
# Parse request body
body = await request.json()
event = body.get("event")
@@ -256,8 +253,6 @@ async def delete_webhook(request: Request) -> JSONResponse:
)
try:
from nextcloud_mcp_server.client.webhooks import WebhooksClient
# Get webhook_id from path parameter
webhook_id = request.path_params.get("webhook_id")
if not webhook_id: