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,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.Contracts] @RESTRICTION EXAMPLES ONLY — this file provides language-specific code patterns. All protocol rules (tier definitions, tag catalog, anchor syntax) are defined exclusively in semantics-core. This file MUST NOT redefine or contradict any rule from semantics-core.

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 the canonical Molecular CoT Logging protocol for belief markers. For the full wire-format specification, see the molecular-cot-logging skill.

ALWAYS import from the shared module — never copy-paste inline:

from ss_tools.lib.cot_logger import log, push_span, pop_span

# Usage:
# log("src_id", "REASON", "intent", payload_dict)
# log("src_id", "EXPLORE", "message", payload_dict, error="assumption violated")
# log("src_id", "REFLECT", "outcome", payload_dict)

Thin context-manager wrappers (backward-compatible aliases for push_span/pop_span):

from contextlib import contextmanager

@contextmanager
def belief_scope(contract_id: str):
    prev_span = push_span(contract_id)
    log(contract_id, "REASON", "enter")
    try:
        yield
    except Exception as e:
        log(contract_id, "EXPLORE", "error", error=str(e))
        raise
    else:
        log(contract_id, "REFLECT", "exit")
    finally:
        pop_span(prev_span)

CRITICAL: All helpers MUST be imported from ss_tools.lib.cot_logger. Never define reason(), explore(), reflect() inline — use the canonical log() function. Do NOT manually type [REASON] in message strings; log() emits the marker field automatically in the molecular-cot JSON wire format.

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:
    log("run_migration_task", "REASON", "Starting migration task", {"task_id": task_id})
    task = await db_session.get(Task, task_id)
    if not task:
        log("run_migration_task", "EXPLORE", "Task not found", error="TaskNotFound")
        raise TaskNotFoundError(task_id)
    try:
        task.status = "RUNNING"
        await db_session.commit()
        log("run_migration_task", "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)
        log("run_migration_task", "REFLECT", "Migration completed successfully", {"task_id": task_id, "dashboards": len(result)})
        return result
    except Exception as e:
        log("run_migration_task", "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:
    log("rebuild_index", "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:
            log("rebuild_index", "EXPLORE", "Parse failure, skipping file", {"file": filepath}, error=str(e))
    snapshot = {"contracts": contracts, "timestamp": datetime.utcnow().isoformat()}
    write_checkpoint(root_path, snapshot)
    log("rebuild_index", "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 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 .

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

V. FASTAPI / ASYNC PATTERNS

Async belief scope

from contextlib import asynccontextmanager
from ss_tools.lib.cot_logger import log, push_span, pop_span

@asynccontextmanager
async def async_belief_scope(contract_id: str):
    prev_span = push_span(contract_id)
    log(contract_id, "REASON", "enter")
    try:
        yield
    except Exception as e:
        log(contract_id, "EXPLORE", "error", error=str(e))
        raise
    else:
        log(contract_id, "REFLECT", "exit")
    finally:
        pop_span(prev_span)

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