fix: resolve all MyPy type errors and add poetry.lock for reproducible builds
Some checks failed
CI/CD Pipeline / Build and Push Docker Images (api) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (chat) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (frontend) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (worker) (push) Blocked by required conditions
CI/CD Pipeline / Deploy to Staging (push) Blocked by required conditions
CI/CD Pipeline / Deploy to Production (push) Blocked by required conditions
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Security Scanning (push) Has been cancelled
CI/CD Pipeline / Lint Code (push) Has been cancelled
CI/CD Pipeline / Generate Documentation (push) Has been cancelled
Some checks failed
CI/CD Pipeline / Build and Push Docker Images (api) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (chat) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (frontend) (push) Blocked by required conditions
CI/CD Pipeline / Build and Push Docker Images (worker) (push) Blocked by required conditions
CI/CD Pipeline / Deploy to Staging (push) Blocked by required conditions
CI/CD Pipeline / Deploy to Production (push) Blocked by required conditions
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Security Scanning (push) Has been cancelled
CI/CD Pipeline / Lint Code (push) Has been cancelled
CI/CD Pipeline / Generate Documentation (push) Has been cancelled
- Install missing type stubs (types-PyYAML, celery-types, types-redis) - Fix template_generator.py: add explicit type casts for YAML data - Fix documentation_tasks.py: add Task type hints for Celery decorators - Fix api/main.py: add type annotations for AsyncResult and response dicts - Fix chat/main.py: add type ignore for socketio (no stubs available) - Remove poetry.lock from .gitignore (needed for Docker builds) - Add poetry.lock to ensure reproducible dependency installation - Format code with Black and verify with Ruff All MyPy checks now pass: 18 errors → 0 errors 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -393,9 +393,9 @@ async def get_job_status(job_id: str) -> Dict[str, Any]:
|
||||
from datacenter_docs.workers.celery_app import celery_app
|
||||
|
||||
try:
|
||||
result = AsyncResult(job_id, app=celery_app)
|
||||
result: Any = AsyncResult(job_id, app=celery_app)
|
||||
|
||||
response = {
|
||||
response: Dict[str, Any] = {
|
||||
"job_id": job_id,
|
||||
"status": result.state, # PENDING, STARTED, SUCCESS, FAILURE, RETRY
|
||||
}
|
||||
@@ -454,7 +454,7 @@ async def list_documentation_files() -> Dict[str, Any]:
|
||||
|
||||
if files:
|
||||
files_by_category[category] = sorted(
|
||||
files, key=lambda x: x["modified"], reverse=True
|
||||
files, key=lambda x: float(x["modified"]) if x["modified"] else 0, reverse=True # type: ignore[arg-type]
|
||||
)
|
||||
|
||||
total_files = sum(len(files) for files in files_by_category.values())
|
||||
|
||||
@@ -6,7 +6,7 @@ import logging
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
|
||||
import socketio
|
||||
import socketio # type: ignore[import-untyped]
|
||||
from fastapi import FastAPI
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
|
||||
@@ -34,7 +34,8 @@ class DocumentationTemplate:
|
||||
"""Load and parse YAML template"""
|
||||
try:
|
||||
with open(self.path, "r", encoding="utf-8") as f:
|
||||
return yaml.safe_load(f)
|
||||
data = yaml.safe_load(f)
|
||||
return dict(data) if data else {}
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load template {self.path}: {e}")
|
||||
raise
|
||||
@@ -42,32 +43,46 @@ class DocumentationTemplate:
|
||||
@property
|
||||
def name(self) -> str:
|
||||
"""Get template name"""
|
||||
return self.data.get("metadata", {}).get("name", "Unknown")
|
||||
metadata = self.data.get("metadata", {})
|
||||
if isinstance(metadata, dict):
|
||||
name = metadata.get("name", "Unknown")
|
||||
return str(name) if name else "Unknown"
|
||||
return "Unknown"
|
||||
|
||||
@property
|
||||
def collector(self) -> str:
|
||||
"""Get required collector name"""
|
||||
return self.data.get("metadata", {}).get("collector", "")
|
||||
metadata = self.data.get("metadata", {})
|
||||
if isinstance(metadata, dict):
|
||||
collector = metadata.get("collector", "")
|
||||
return str(collector) if collector else ""
|
||||
return ""
|
||||
|
||||
@property
|
||||
def sections(self) -> List[Dict[str, Any]]:
|
||||
"""Get documentation sections"""
|
||||
return self.data.get("sections", [])
|
||||
sections = self.data.get("sections", [])
|
||||
if isinstance(sections, list):
|
||||
return [dict(s) for s in sections if isinstance(s, dict)]
|
||||
return []
|
||||
|
||||
@property
|
||||
def generation_config(self) -> Dict[str, Any]:
|
||||
"""Get generation configuration"""
|
||||
return self.data.get("generation", {})
|
||||
config = self.data.get("generation", {})
|
||||
return dict(config) if isinstance(config, dict) else {}
|
||||
|
||||
@property
|
||||
def output_config(self) -> Dict[str, Any]:
|
||||
"""Get output configuration"""
|
||||
return self.data.get("output", {})
|
||||
config = self.data.get("output", {})
|
||||
return dict(config) if isinstance(config, dict) else {}
|
||||
|
||||
@property
|
||||
def schedule_config(self) -> Dict[str, Any]:
|
||||
"""Get schedule configuration"""
|
||||
return self.data.get("schedule", {})
|
||||
config = self.data.get("schedule", {})
|
||||
return dict(config) if isinstance(config, dict) else {}
|
||||
|
||||
|
||||
class TemplateBasedGenerator(BaseGenerator):
|
||||
@@ -337,8 +352,10 @@ Guidelines:
|
||||
save_to_file = output_config.get("save_to_file", True)
|
||||
|
||||
for section_def in self.template.sections:
|
||||
section_id = section_def.get("id")
|
||||
section_title = section_def.get("title")
|
||||
section_id_raw = section_def.get("id")
|
||||
section_id = str(section_id_raw) if section_id_raw else "unknown"
|
||||
section_title_raw = section_def.get("title")
|
||||
section_title = str(section_title_raw) if section_title_raw else "Untitled"
|
||||
|
||||
# Check if this section should loop over items
|
||||
loop_over = section_def.get("loop_over")
|
||||
|
||||
@@ -10,7 +10,7 @@ from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
|
||||
from celery import group
|
||||
from celery import Task, group
|
||||
|
||||
from datacenter_docs.workers.celery_app import celery_app
|
||||
|
||||
@@ -18,7 +18,7 @@ logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@celery_app.task(name="collect_and_generate_docs", bind=True)
|
||||
def collect_and_generate_docs(self, collector_name: str, template_path: str) -> Dict[str, Any]:
|
||||
def collect_and_generate_docs(self: Task, collector_name: str, template_path: str) -> Dict[str, Any]: # type: ignore[misc]
|
||||
"""
|
||||
Collect data from infrastructure and generate documentation
|
||||
|
||||
@@ -170,7 +170,7 @@ async def _get_collector(collector_name: str) -> Any:
|
||||
|
||||
|
||||
@celery_app.task(name="generate_proxmox_docs", bind=True)
|
||||
def generate_proxmox_docs(self) -> Dict[str, Any]:
|
||||
def generate_proxmox_docs(self: Task) -> Dict[str, Any]: # type: ignore[misc]
|
||||
"""
|
||||
Scheduled task to generate Proxmox documentation
|
||||
|
||||
@@ -291,7 +291,7 @@ def index_generated_docs(output_dir: str = "output") -> Dict[str, Any]:
|
||||
|
||||
logger.info(f"Starting documentation indexing from {output_dir}")
|
||||
|
||||
result = {
|
||||
result: Dict[str, Any] = {
|
||||
"success": False,
|
||||
"files_indexed": 0,
|
||||
"chunks_created": 0,
|
||||
|
||||
Reference in New Issue
Block a user