---
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]

## 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:**

```python
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`):

```python
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
```python
# #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
```python
# #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
```python
# #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
```python
# #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
```python
# #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
```python
# #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
```python
# #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

```bash
# 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
```python
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