ss-tools/.opencode/skills/semantics-python/SKILL.md

name: semantics-python description: Python-specific GRACE-Poly protocol: few-shot complexity examples, belief runtime patterns, module conventions, and FastAPI/SQLAlchemy patterns for ss-tools.

#region Std.Semantics.Python [C:4] [TYPE Skill] [SEMANTICS python,examples,belief,fastapi,sqlalchemy] @BRIEF Python-specific HOW: few-shot complexity examples, belief runtime patterns, module decomposition, and FastAPI/SQLAlchemy conventions for the GRACE-Poly protocol in ss-tools. @RELATION DEPENDS_ON -> [Std.Semantics.Core] @RELATION DEPENDS_ON -> [Std.Semantics.Belief] @RELATION DEPENDS_ON -> [Std.Semantics.Contracts]

0. WHEN TO USE THIS SKILL

Load this skill when implementing Python backend code under the GRACE-Poly protocol in ss-tools. It provides concrete Python examples for each complexity tier, belief runtime patterns, FastAPI/SQLAlchemy conventions, and module structure rules. For generic protocol rules, see semantics-core. For contract enforcement methodology, see semantics-contracts.

I. PYTHON BELIEF RUNTIME PATTERNS

ss-tools uses structured JSON logging for belief markers. The canonical pattern:

import logging
import json
from contextlib import contextmanager

logger = logging.getLogger(__name__)

@contextmanager
def belief_scope(contract_id: str):
    """Thread-local belief frame for C4/C5 functions."""
    logger.info(json.dumps({"marker": "REASON", "contract": contract_id, "event": "enter"}))
    try:
        yield
    except Exception as e:
        logger.error(json.dumps({"marker": "EXPLORE", "contract": contract_id, "error": str(e)}))
        raise
    else:
        logger.info(json.dumps({"marker": "REFLECT", "contract": contract_id, "event": "exit"}))

# Usage:
def reason(message: str, extra: dict = None):
    logger.info(json.dumps({"marker": "REASON", "message": message, **(extra or {})}))

def explore(message: str, extra: dict = None):
    logger.warning(json.dumps({"marker": "EXPLORE", "message": message, **(extra or {})}))

def reflect(message: str, extra: dict = None):
    logger.info(json.dumps({"marker": "REFLECT", "message": message, **(extra or {})}))

CRITICAL: Do NOT manually type [REASON] in message strings. ALWAYS pass structured data through the extra dict. The marker field is the canonical protocol marker.

II. PYTHON COMPLEXITY EXAMPLES

C1 (Atomic) — DTOs, Pydantic schemas, simple constants

# #region UserResponseSchema [C:1] [TYPE Class]
from pydantic import BaseModel

class UserResponseSchema(BaseModel):
    id: str
    username: str
    email: str
# #endregion UserResponseSchema

C2 (Simple) — Pure functions, utility helpers

# #region format_timestamp [C:2] [TYPE Function] [SEMANTICS time,formatting]
# @BRIEF Format a UTC datetime into a human-readable ISO-8601 string.
from datetime import datetime

def format_timestamp(ts: datetime) -> str:
    return ts.strftime("%Y-%m-%dT%H:%M:%SZ")
# #endregion format_timestamp

C3 (Flow) — Module with nested functions, service layer

# #region dashboard_migration [C:3] [TYPE Module] [SEMANTICS migration,dashboard]
# @BRIEF Dashboard migration service — export/import dashboards with validation.
# @LAYER Service

# #region migrate_dashboard [C:3] [TYPE Function] [SEMANTICS migration,dashboard]
# @BRIEF Migrate a single dashboard from source to target Superset instance.
# @RELATION DEPENDS_ON -> [SupersetClient]
# @RELATION DEPENDS_ON -> [DashboardValidator]
def migrate_dashboard(source_client, target_client, dashboard_id: str, db_mapping: dict) -> dict:
    dashboard = source_client.get_dashboard(dashboard_id)
    validate_dashboard(dashboard)
    mapped = apply_db_mapping(dashboard, db_mapping)
    result = target_client.import_dashboard(mapped)
    return result
# #endregion migrate_dashboard

# #endregion dashboard_migration

C4 (Orchestration) — Stateful operations with belief runtime

# #region run_migration_task [C:4] [TYPE Function] [SEMANTICS migration,task,state]
# @BRIEF Execute a full migration task with rollback capability and progress reporting.
# @PRE Database connection is established. Task record exists with valid migration plan.
# @POST Task status updated to COMPLETED or FAILED. Migration audit log written.
# @SIDE_EFFECT Modifies target Superset instance; writes task progress to DB; sends WebSocket updates.
# @RELATION DEPENDS_ON -> [TaskManager]
# @RELATION DEPENDS_ON -> [MigrationService]
# @RELATION DEPENDS_ON -> [WebSocketNotifier]
async def run_migration_task(task_id: str, db_session) -> dict:
    reason("Starting migration task", {"task_id": task_id})
    task = await db_session.get(Task, task_id)
    if not task:
        explore("Task not found", {"task_id": task_id})
        raise TaskNotFoundError(task_id)
    try:
        task.status = "RUNNING"
        await db_session.commit()
        reason("Task status set to RUNNING", {"task_id": task_id})
        result = await execute_migration_plan(task.migration_plan)
        task.status = "COMPLETED"
        task.result = result
        await db_session.commit()
        await notify_frontend(task_id, "completed", result)
        reflect("Migration completed successfully", {"task_id": task_id, "dashboards": len(result)})
        return result
    except Exception as e:
        explore("Migration failed, rolling back", {"task_id": task_id, "error": str(e)})
        task.status = "FAILED"
        task.error = str(e)
        await db_session.commit()
        await notify_frontend(task_id, "failed", {"error": str(e)})
        raise
# #endregion run_migration_task

C5 (Critical) — With decision memory

# #region rebuild_index [C:5] [TYPE Function] [SEMANTICS indexing,recovery,semantic]
# @BRIEF Rebuild the full semantic index from source with atomic swap and rollback.
# @PRE Workspace root is accessible. Source files exist.
# @POST New index atomically swapped; old preserved for rollback.
# @SIDE_EFFECT Reads all source files; writes index snapshot and checkpoint metadata.
# @DATA_CONTRACT Input: WorkspaceRoot -> Output: IndexSnapshot + CheckpointManifest
# @INVARIANT Index consistency: every contract_id in edges maps to an existing node.
# @RELATION DEPENDS_ON -> [FileScanner]
# @RELATION DEPENDS_ON -> [ContractParser]
# @RELATION DEPENDS_ON -> [CheckpointWriter]
# @RATIONALE Full rebuild needed because incremental update cannot detect deleted contracts.
# @REJECTED Incremental-only update was rejected — it leaves stale edges when contracts
#           are deleted; only full scan guarantees consistency.
def rebuild_index(root_path: str) -> dict:
    reason("Scanning source files", {"root": root_path})
    contracts = []
    for filepath in scan_files(root_path):
        try:
            parsed = parse_contract(filepath)
            contracts.append(parsed)
        except Exception as e:
            explore("Parse failure, skipping file", {"file": filepath, "error": str(e)})
    snapshot = {"contracts": contracts, "timestamp": datetime.utcnow().isoformat()}
    write_checkpoint(root_path, snapshot)
    reflect("Rebuild complete", {"contracts": len(contracts)})
    return snapshot
# #endregion rebuild_index

III. PYTHON MODULE PATTERNS

Project module layout (ss-tools convention)

backend/
├── src/
│   ├── api/               # FastAPI route handlers (C3)
│   ├── core/              # Business logic core (C4/C5)
│   │   ├── task_manager/  # Async task orchestration
│   │   ├── auth/          # Authentication/authorization
│   │   ├── migration/     # Dashboard migration logic
│   │   └── plugins/       # Plugin system
│   ├── models/            # SQLAlchemy models (C1/C2)
│   ├── services/          # Business-logic services (C3/C4)
│   └── schemas/           # Pydantic request/response schemas (C1)
└── tests/                 # pytest test modules

Module decomposition rules

• Module files MUST stay < 400 LOC
• Individual contract nodes MUST stay < 150 LOC, Cyclomatic Complexity ≤ 10
• When limits are breached: extract into new modules with @RELATION edges
• Use __init__.py for public re-exports only, not for logic
• FastAPI route modules: one file per resource group (e.g., dashboards.py, datasets.py)

Comment style

• Python: # #region ... / # #endregion ...
• Docstrings for metadata: @TAG: on separate lines within the region
• Legacy [DEF:...] / [/DEF:...] recognized but new code uses region format

FastAPI route pattern

# #region dashboard_routes [C:3] [TYPE Module] [SEMANTICS api,dashboard]
# @BRIEF Dashboard CRUD and migration API routes.
# @RELATION DEPENDS_ON -> [DashboardService]
# @RELATION DEPENDS_ON -> [AuthMiddleware]
from fastapi import APIRouter, Depends

router = APIRouter(prefix="/api/dashboards", tags=["dashboards"])

# #region list_dashboards [C:2] [TYPE Function] [SEMANTICS api,query]
# @BRIEF List dashboards with optional filters.
@router.get("/")
async def list_dashboards(
    page: int = 1,
    page_size: int = 20,
    service=Depends(get_dashboard_service)
):
    return await service.list_dashboards(page, page_size)
# #endregion list_dashboards

# #endregion dashboard_routes

SQLAlchemy model pattern

# #region Dashboard [C:1] [TYPE Class]
from sqlalchemy import Column, String, DateTime, JSON
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Dashboard(Base):
    __tablename__ = "dashboards"
    id = Column(String, primary_key=True)
    title = Column(String, nullable=False)
    metadata = Column(JSON)
    created_at = Column(DateTime, server_default="now()")
# #endregion Dashboard

IV. PYTHON VERIFICATION

# Backend tests (from backend/ directory)
cd backend && source .venv/bin/activate && python -m pytest -v

# With coverage
python -m pytest --cov=src --cov-report=term-missing

# Ruff linting
python -m ruff check src/ tests/

# Type checking (if mypy is configured)
python -m mypy src/

V. FASTAPI / ASYNC PATTERNS

Async belief scope

import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def async_belief_scope(contract_id: str):
    """Async belief frame for C4/C5 async functions."""
    reason("enter", {"contract": contract_id})
    try:
        yield
    except Exception as e:
        explore("error", {"contract": contract_id, "error": str(e)})
        raise
    else:
        reflect("exit", {"contract": contract_id})

Dependency injection convention

• Use FastAPI Depends() for injecting services
• Services are singletons or request-scoped
• Never use global mutable state in service layer

#endregion Std.Semantics.Python