will/unitforge
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

backend: remove __main__ runner; typing: fix mypy across core and routes; tooling: use uv for pytest in pre-commit; security: narrow broad except in config; mypy: relax untyped decorator check for backend.app.main

Browse Source
This commit is contained in:
William Valentin
2025-09-15 01:25:25 -07:00
parent be520c14e9
commit b544399f9c
9 changed files with 69 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/core/config.py
+1 -1
View File
@@ -207,7 +207,7 @@ class Settings:
parsed = self._parse_list("CORS_ORIGINS", [])
if parsed:
return parsed
except Exception:
except (ValueError, SyntaxError, json.JSONDecodeError, TypeError):
# Fall back to comma-separated parsing if JSON/literal parsing fails
pass
backend/app/core/templates.py
+10 -10
View File
@@ -43,7 +43,7 @@ class UnitTemplate:
class WebApplicationTemplate(UnitTemplate):
"""Template for web application services."""
def __init__(self):
def __init__(self) -> None:
super().__init__(
name="webapp",
description="Web application service (Node.js, Python, etc.)",
@@ -159,7 +159,7 @@ class WebApplicationTemplate(UnitTemplate):
class DatabaseTemplate(UnitTemplate):
"""Template for database services."""
def __init__(self):
def __init__(self) -> None:
super().__init__(
name="database",
description="Database service (PostgreSQL, MySQL, MongoDB, etc.)",
@@ -249,7 +249,7 @@ class DatabaseTemplate(UnitTemplate):
class BackupTimerTemplate(UnitTemplate):
"""Template for backup timer services."""
def __init__(self):
def __init__(self) -> None:
super().__init__(
name="backup-timer",
description="Scheduled backup service with timer",
@@ -334,7 +334,7 @@ class BackupTimerTemplate(UnitTemplate):
class ProxySocketTemplate(UnitTemplate):
"""Template for socket-activated proxy services."""
def __init__(self):
def __init__(self) -> None:
super().__init__(
name="proxy-socket",
description="Socket-activated proxy service",
@@ -436,7 +436,7 @@ class ProxySocketTemplate(UnitTemplate):
class ContainerTemplate(UnitTemplate):
"""Template for containerized services (Docker/Podman)."""
def __init__(self):
def __init__(self) -> None:
super().__init__(
name="container",
description="Containerized service (Docker/Podman)",
@@ -596,11 +596,11 @@ class ContainerTemplate(UnitTemplate):
class TemplateRegistry:
"""Registry for managing available unit file templates."""
def __init__(self):
self._templates = {}
def __init__(self) -> None:
self._templates: Dict[str, UnitTemplate] = {}
self._register_default_templates()
def _register_default_templates(self):
def _register_default_templates(self) -> None:
"""Register all default templates."""
templates = [
WebApplicationTemplate(),
@@ -613,7 +613,7 @@ class TemplateRegistry:
for template in templates:
self.register(template)
def register(self, template: UnitTemplate):
def register(self, template: UnitTemplate) -> None:
"""Register a new template."""
self._templates[template.name] = template
@@ -636,7 +636,7 @@ class TemplateRegistry:
def search(self, query: str) -> List[UnitTemplate]:
"""Search templates by name, description, or tags."""
query = query.lower()
results = []
results: List[UnitTemplate] = []
for template in self._templates.values():
# Search in name and description
backend/app/core/unit_file.py
+23 -16
View File
@@ -9,7 +9,7 @@ import configparser
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Optional
from typing import Any, Dict, List, Optional
class UnitType(Enum):
@@ -46,6 +46,11 @@ class UnitFileInfo:
conflicts: List[str] = field(default_factory=list)
class CaseConfigParser(configparser.ConfigParser):
def optionxform(self, optionstr: str) -> str:
return str(optionstr)
class SystemdUnitFile:
"""
Parser and validator for systemd unit files.
@@ -306,7 +311,7 @@ class SystemdUnitFile:
}
# Service types and their requirements
SERVICE_TYPES = {
SERVICE_TYPES: Dict[str, Dict[str, List[str]]] = {
"simple": {"required": ["ExecStart"], "conflicts": ["BusName", "Type=forking"]},
"exec": {"required": ["ExecStart"], "conflicts": ["BusName"]},
"forking": {"recommended": ["PIDFile"], "conflicts": ["BusName"]},
@@ -320,11 +325,10 @@ class SystemdUnitFile:
"""Initialize with either content string or file path."""
self.content = content or ""
self.file_path = file_path
self.config = configparser.ConfigParser(
self.config = CaseConfigParser(
interpolation=None, allow_no_value=True, delimiters=("=",)
)
self.config.optionxform = lambda optionstr: str(optionstr) # Preserve case
self._parse_errors = []
self._parse_errors: List[ValidationError] = []
if content:
self._parse_content(content)
@@ -419,7 +423,7 @@ class SystemdUnitFile:
def validate(self) -> List[ValidationError]:
"""Validate the unit file and return list of errors/warnings."""
errors = self._parse_errors.copy()
errors: List[ValidationError] = self._parse_errors.copy()
# Check for basic structure
if not self.config.sections():
@@ -443,7 +447,7 @@ class SystemdUnitFile:
def _validate_section(self, section: str) -> List[ValidationError]:
"""Validate a specific section."""
errors = []
errors: List[ValidationError] = []
# Check if section is known
if section not in self.COMMON_SECTIONS:
@@ -477,7 +481,7 @@ class SystemdUnitFile:
self, section: str, key: str, value: str
) -> List[ValidationError]:
"""Validate specific key-value pairs."""
errors = []
errors: List[ValidationError] = []
# Service-specific validations
if section == "Service":
@@ -574,7 +578,7 @@ class SystemdUnitFile:
def _validate_unit_type(self, unit_type: UnitType) -> List[ValidationError]:
"""Perform type-specific validation."""
errors = []
errors: List[ValidationError] = []
if unit_type == UnitType.SERVICE:
errors.extend(self._validate_service())
@@ -589,13 +593,14 @@ class SystemdUnitFile:
def _validate_service(self) -> List[ValidationError]:
"""Validate service-specific requirements."""
errors = []
errors: List[ValidationError] = []
if not self.config.has_section("Service"):
return errors
service_type = self.config.get("Service", "Type", fallback="simple")
type_config = self.SERVICE_TYPES.get(service_type, {})
tmp = self.SERVICE_TYPES.get(service_type)
type_config: Dict[str, List[str]] = {} if tmp is None else tmp
# Check required keys
for required_key in type_config.get("required", []):
@@ -648,7 +653,7 @@ class SystemdUnitFile:
def _validate_timer(self) -> List[ValidationError]:
"""Validate timer-specific requirements."""
errors = []
errors: List[ValidationError] = []
if not self.config.has_section("Timer"):
return errors
@@ -678,7 +683,7 @@ class SystemdUnitFile:
def _validate_socket(self) -> List[ValidationError]:
"""Validate socket-specific requirements."""
errors = []
errors: List[ValidationError] = []
if not self.config.has_section("Socket"):
return errors
@@ -709,7 +714,7 @@ class SystemdUnitFile:
def _validate_mount(self) -> List[ValidationError]:
"""Validate mount-specific requirements."""
errors = []
errors: List[ValidationError] = []
if not self.config.has_section("Mount"):
return errors
@@ -755,7 +760,9 @@ class SystemdUnitFile:
for section in self.config.sections():
lines.append(f"[{section}]")
for key in self.config.options(section):
value = self.config.get(section, key)
from typing import cast
value = cast(Optional[str], self.config.get(section, key))
if value is None:
lines.append(key)
else:
@@ -796,7 +803,7 @@ class SystemdUnitFile:
return self.config.options(section)
def create_unit_file(unit_type: UnitType, **kwargs) -> SystemdUnitFile:
def create_unit_file(unit_type: UnitType, **kwargs: Any) -> SystemdUnitFile:
"""
Create a new unit file of the specified type with basic structure.
backend/app/main.py
+23 -28
View File
@@ -9,12 +9,12 @@ import tempfile
from pathlib import Path
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, File, HTTPException, Request, UploadFile # type: ignore
from fastapi.middleware.cors import CORSMiddleware # type: ignore
from fastapi.middleware.gzip import GZipMiddleware # type: ignore
from fastapi.responses import FileResponse, HTMLResponse # type: ignore
from fastapi.staticfiles import StaticFiles # type: ignore
from fastapi.templating import Jinja2Templates # type: ignore
from fastapi import FastAPI, File, HTTPException, Request, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
from .core.config import settings
@@ -137,7 +137,7 @@ def template_to_dict(template: UnitTemplate) -> Dict[str, Any]:
# Web UI Routes
@app.get("/", response_class=HTMLResponse)
async def index(request: Request):
async def index(request: Request) -> Any:
"""Serve the main web interface."""
context = {"request": request}
context.update(settings.get_template_context())
@@ -145,7 +145,7 @@ async def index(request: Request):
@app.get("/editor", response_class=HTMLResponse)
async def editor(request: Request):
async def editor(request: Request) -> Any:
"""Serve the unit file editor interface."""
context = {"request": request}
context.update(settings.get_template_context())
@@ -153,15 +153,16 @@ async def editor(request: Request):
@app.get("/templates", response_class=HTMLResponse)
async def templates_page(request: Request):
async def templates_page(request: Request) -> Any:
"""Serve the templates browser interface."""
context = {"request": request}
context.update(settings.get_template_context())
return templates.TemplateResponse("templates.html", context)
# CLI Info Page
@app.get("/cli", response_class=HTMLResponse)
async def cli_page(request: Request):
async def cli_page(request: Request) -> Any:
"""Serve the CLI information page."""
context = {"request": request}
context.update(settings.get_template_context())
@@ -170,7 +171,7 @@ async def cli_page(request: Request):
# API Routes
@app.post("/api/validate", response_model=ValidationResult)
async def validate_unit_file(unit_file: UnitFileContent):
async def validate_unit_file(unit_file: UnitFileContent) -> ValidationResult:
"""Validate a systemd unit file."""
try:
systemd_unit = SystemdUnitFile(content=unit_file.content)
@@ -194,7 +195,7 @@ async def validate_unit_file(unit_file: UnitFileContent):
@app.post("/api/generate")
async def generate_unit_file(request: GenerateRequest):
async def generate_unit_file(request: GenerateRequest) -> Dict[str, Any]:
"""Generate a unit file from a template."""
try:
template = template_registry.get_template(request.template_name)
@@ -246,7 +247,7 @@ async def generate_unit_file(request: GenerateRequest):
@app.post("/api/create")
async def create_unit_file_endpoint(request: CreateUnitRequest):
async def create_unit_file_endpoint(request: CreateUnitRequest) -> Dict[str, Any]:
"""Create a basic unit file."""
try:
# Parse unit type
@@ -293,14 +294,14 @@ async def create_unit_file_endpoint(request: CreateUnitRequest):
@app.get("/api/templates", response_model=List[TemplateInfo])
async def list_templates():
async def list_templates() -> List[TemplateInfo]:
"""List all available templates."""
templates = template_registry.list_templates()
return [TemplateInfo(**template_to_dict(template)) for template in templates]
@app.get("/api/templates/{template_name}", response_model=TemplateInfo)
async def get_template(template_name: str):
async def get_template(template_name: str) -> TemplateInfo:
"""Get details of a specific template."""
template = template_registry.get_template(template_name)
if not template:
@@ -312,14 +313,14 @@ async def get_template(template_name: str):
@app.get("/api/templates/category/{category}", response_model=List[TemplateInfo])
async def get_templates_by_category(category: str):
async def get_templates_by_category(category: str) -> List[TemplateInfo]:
"""Get templates by category."""
templates = template_registry.get_by_category(category)
return [TemplateInfo(**template_to_dict(template)) for template in templates]
@app.get("/api/templates/type/{unit_type}", response_model=List[TemplateInfo])
async def get_templates_by_type(unit_type: str):
async def get_templates_by_type(unit_type: str) -> List[TemplateInfo]:
"""Get templates by unit type."""
try:
unit_type_enum = UnitType(unit_type.lower())
@@ -331,14 +332,14 @@ async def get_templates_by_type(unit_type: str):
@app.get("/api/search/{query}", response_model=List[TemplateInfo])
async def search_templates(query: str):
async def search_templates(query: str) -> List[TemplateInfo]:
"""Search templates by name, description, or tags."""
templates = template_registry.search(query)
return [TemplateInfo(**template_to_dict(template)) for template in templates]
@app.post("/api/download")
async def download_unit_file(unit_file: UnitFileContent):
async def download_unit_file(unit_file: UnitFileContent) -> FileResponse:
"""Download a unit file."""
try:
# Create a temporary file
@@ -359,7 +360,7 @@ async def download_unit_file(unit_file: UnitFileContent):
@app.post("/api/upload")
async def upload_unit_file(file: UploadFile = File(...)):
async def upload_unit_file(file: UploadFile = File(...)) -> Dict[str, Any]:
"""Upload and validate a unit file."""
try:
# Check file size
@@ -424,7 +425,7 @@ async def upload_unit_file(file: UploadFile = File(...)):
@app.get("/api/info")
async def get_info():
async def get_info() -> Dict[str, Any]:
"""Get application information."""
return {
"name": settings.app_name,
@@ -449,7 +450,7 @@ async def get_info():
# Health check
@app.get("/health")
async def health_check():
async def health_check() -> Dict[str, Any]:
"""Health check endpoint."""
if not settings.health_check_enabled:
raise HTTPException(status_code=404, detail="Health check disabled")
@@ -461,9 +462,3 @@ async def health_check():
"environment": settings.environment,
"timestamp": __import__("datetime").datetime.utcnow().isoformat(),
}
if __name__ == "__main__":
import uvicorn # type: ignore
uvicorn.run(app, host=settings.host, port=settings.port) # nosec B104