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

fix: resolve type checking warnings for CI

Browse Source 
- Add type casts for Starlette app state access
- Add assertions for cipher, card, board, stack after initialization
- Add None checks for XML element text attributes
- Handle __package__ being None in tracing setup
- Fix TokenBrokerService initialization to use storage credentials

Resolves 42 type warnings from ty-check, enabling CI linting to pass.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2025-12-18 00:44:58 +01:00
parent 54b69f0d68
commit e4f3beee01
8 changed files with 49 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/app.py
+15 -11
View File
@@ -5,7 +5,7 @@ from collections.abc import AsyncIterator
from contextlib import AsyncExitStack, asynccontextmanager
from contextvars import ContextVar
from dataclasses import dataclass
from typing import TYPE_CHECKING, Optional
from typing import TYPE_CHECKING, Optional, cast
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
@@ -1259,7 +1259,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
# We need to find it in the mounted routes
for route in app.routes:
if isinstance(route, Mount) and route.path == "/app":
route.app.state.oauth_context = oauth_context_dict
browser_app = cast(Starlette, route.app)
browser_app.state.oauth_context = oauth_context_dict
logger.info(
"OAuth context shared with browser_app for session auth"
)
@@ -1280,7 +1281,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
# Also share with browser_app for webhook routes
for route in app.routes:
if isinstance(route, Mount) and route.path == "/app":
route.app.state.storage = storage
browser_app = cast(Starlette, route.app)
browser_app.state.storage = storage
logger.info(
"Storage shared with browser_app for webhook management"
)
@@ -1348,10 +1350,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
# Also share with browser_app for /app route
for route in app.routes:
if isinstance(route, Mount) and route.path == "/app":
route.app.state.document_send_stream = send_stream
route.app.state.document_receive_stream = receive_stream
route.app.state.shutdown_event = shutdown_event
route.app.state.scanner_wake_event = scanner_wake_event
browser_app = cast(Starlette, route.app)
browser_app.state.document_send_stream = send_stream
browser_app.state.document_receive_stream = receive_stream
browser_app.state.shutdown_event = shutdown_event
browser_app.state.scanner_wake_event = scanner_wake_event
logger.info("Vector sync state shared with browser_app for /app")
break
@@ -1481,10 +1484,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
# Also share with browser_app for /app route
for route in app.routes:
if isinstance(route, Mount) and route.path == "/app":
route.app.state.document_send_stream = send_stream
route.app.state.document_receive_stream = receive_stream
route.app.state.shutdown_event = shutdown_event
route.app.state.scanner_wake_event = scanner_wake_event
browser_app = cast(Starlette, route.app)
browser_app.state.document_send_stream = send_stream
browser_app.state.document_receive_stream = receive_stream
browser_app.state.shutdown_event = shutdown_event
browser_app.state.scanner_wake_event = scanner_wake_event
logger.info("Vector sync state shared with browser_app for /app")
break
nextcloud_mcp_server/auth/storage.py
+14
View File
@@ -216,6 +216,8 @@ class RefreshTokenStorage:
if not self._initialized:
await self.initialize()
# Type narrowing: cipher is set after initialize()
assert self.cipher is not None
encrypted_token = self.cipher.encrypt(refresh_token.encode())
now = int(time.time())
scopes_json = json.dumps(scopes) if scopes else None
@@ -361,6 +363,9 @@ class RefreshTokenStorage:
if not self._initialized:
await self.initialize()
# Type narrowing: cipher is set after initialize()
assert self.cipher is not None
start_time = time.time()
try:
async with aiosqlite.connect(self.db_path) as db:
@@ -445,6 +450,9 @@ class RefreshTokenStorage:
if not self._initialized:
await self.initialize()
# Type narrowing: cipher is set after initialize()
assert self.cipher is not None
async with aiosqlite.connect(self.db_path) as db:
async with db.execute(
"""
@@ -616,6 +624,9 @@ class RefreshTokenStorage:
if not self._initialized:
await self.initialize()
# Type narrowing: cipher is set after initialize()
assert self.cipher is not None
# Encrypt sensitive data
encrypted_secret = self.cipher.encrypt(client_secret.encode())
encrypted_reg_token = (
@@ -686,6 +697,9 @@ class RefreshTokenStorage:
if not self._initialized:
await self.initialize()
# Type narrowing: cipher is set after initialize()
assert self.cipher is not None
async with aiosqlite.connect(self.db_path) as db:
async with db.execute(
"""
nextcloud_mcp_server/client/webdav.py
+2
View File
@@ -1180,9 +1180,11 @@ class WebDAVClient(BaseNextcloudClient):
"name": display_name_elem.text,
"userVisible": user_visible_elem.text.lower() == "true"
if user_visible_elem is not None
and user_visible_elem.text is not None
else True,
"userAssignable": user_assignable_elem.text.lower() == "true"
if user_assignable_elem is not None
and user_assignable_elem.text is not None
else True,
}
logger.debug(f"Found tag '{tag_name}' with ID {tag_info['id']}")
nextcloud_mcp_server/observability/tracing.py
+2 -1
View File
@@ -53,10 +53,11 @@ def setup_tracing(
global _tracer
# Create resource with service name
pkg_name = __package__.split(".")[0] if __package__ else "nextcloud_mcp_server"
resource = Resource.create(
{
"service.name": service_name,
"service.version": version(__package__.split(".")[0]),
"service.version": version(pkg_name),
}
)
nextcloud_mcp_server/search/context.py
+3
View File
@@ -665,6 +665,9 @@ async def _fetch_document_text(
logger.warning(f"Deck card {doc_id} not found in any board/stack")
return None
# Type narrowing: card is set if we reach here
assert card is not None
# Reconstruct full content as indexed: title + "\n\n" + description
# This ensures chunk offsets align with indexed content structure
content_parts = [card.title]
nextcloud_mcp_server/server/oauth_tools.py
+6 -4
View File
@@ -418,11 +418,12 @@ async def revoke_nextcloud_access(
storage = RefreshTokenStorage.from_env()
await storage.initialize()
encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY")
if not encryption_key:
# Get OAuth client credentials from storage
client_creds = await storage.get_oauth_client()
if not client_creds:
return RevocationResult(
success=False,
message="Token encryption key not configured.",
message="OAuth client credentials not found in storage.",
)
broker = TokenBrokerService(
@@ -432,7 +433,8 @@ async def revoke_nextcloud_access(
f"{os.getenv('NEXTCLOUD_HOST')}/.well-known/openid-configuration",
),
nextcloud_host=os.getenv("NEXTCLOUD_HOST"), # type: ignore
encryption_key=encryption_key,
client_id=client_creds["client_id"],
client_secret=client_creds["client_secret"],
)
# Revoke access
nextcloud_mcp_server/vector/processor.py
+5
View File
@@ -379,6 +379,11 @@ async def _index_document(
f"Deck card {doc_task.doc_id} not found in any board/stack"
)
# Type narrowing: card, board, stack are all set if we reach here
assert card is not None
assert board is not None
assert stack is not None
# Build content from card title and description
content_parts = [card.title]
if card.description:
nextcloud_mcp_server/vector/qdrant_client.py
+2
View File
@@ -89,6 +89,8 @@ async def get_qdrant_client() -> AsyncQdrantClient:
if isinstance(vectors, dict):
actual_dimension = vectors["dense"].size
else:
# Type narrowing: vectors must be VectorParams if not dict
assert isinstance(vectors, VectorParams)
actual_dimension = vectors.size
# Validate dimension matches