285 lines
12 KiB
Markdown
285 lines
12 KiB
Markdown
---
|
|
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]
|
|
@RELATION DISPATCHES -> [MolecularCoTLogging]
|
|
@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`.
|
|
@RATIONALE Python's async/await model, FastAPI dependency injection, and SQLAlchemy session management create unique failure modes for Transformer agents: (1) async/await boundary confusion — agents write sync code in async contexts or forget `await` on ORM calls, producing silent no-ops; (2) dependency injection blindness — FastAPI's `Depends()` creates implicit call graphs that the agent's attention cannot trace without explicit @RELATION edges; (3) session lifecycle drift — SQLAlchemy sessions have strict boundaries that agents violate by passing detached objects across function calls. Concrete examples at each complexity tier act as few-shot anchors that override the agent's pre-trained (and often wrong) Python patterns.
|
|
@REJECTED Generic Python patterns without GRACE anchors were rejected — agents produce working code that violates module size limits (INV_7), omits belief runtime markers, and creates orphan contracts invisible to the semantic index. Relying on the agent's pre-trained FastAPI/SQLAlchemy knowledge without project-specific examples was rejected — ss-tools has specific conventions (trace_id propagation, plugin architecture, WebSocket logging) that general training data cannot capture.
|
|
|
|
## 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 Users.UserResponseSchema [C:1] [TYPE Class]
|
|
from pydantic import BaseModel
|
|
|
|
class UserResponseSchema(BaseModel):
|
|
id: str
|
|
username: str
|
|
email: str
|
|
# #endregion Users.UserResponseSchema
|
|
```
|
|
|
|
### C2 (Simple) — Pure functions, utility helpers
|
|
```python
|
|
# #region Time.FormatTimestamp [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 Time.FormatTimestamp
|
|
```
|
|
|
|
### C3 (Flow) — Module with nested functions, service layer
|
|
```python
|
|
# #region Migration.Dashboard [C:3] [TYPE Module] [SEMANTICS migration,dashboard]
|
|
# @BRIEF Dashboard migration service — export/import dashboards with validation.
|
|
# @LAYER Service
|
|
|
|
# #region Migration.Dashboard.Migrate [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 Migration.Dashboard.Migrate
|
|
|
|
# #endregion Migration.Dashboard
|
|
```
|
|
|
|
### C4 (Orchestration) — Stateful operations with belief runtime
|
|
```python
|
|
# #region Migration.RunTask [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("Migration.RunTask", "REASON", "Starting migration task", {"task_id": task_id})
|
|
task = await db_session.get(Task, task_id)
|
|
if not task:
|
|
log("Migration.RunTask", "EXPLORE", "Task not found", error="TaskNotFound")
|
|
raise TaskNotFoundError(task_id)
|
|
try:
|
|
task.status = "RUNNING"
|
|
await db_session.commit()
|
|
log("Migration.RunTask", "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("Migration.RunTask", "REFLECT", "Migration completed", {"task_id": task_id, "dashboards": len(result)})
|
|
return result
|
|
except Exception as e:
|
|
log("Migration.RunTask", "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 Migration.RunTask
|
|
```
|
|
|
|
### C5 (Critical) — With decision memory
|
|
```python
|
|
# #region Index.Rebuild [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("Index.Rebuild", "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("Index.Rebuild", "EXPLORE", "Parse failure, skipping file", {"file": filepath}, error=str(e))
|
|
snapshot = {"contracts": contracts, "timestamp": datetime.utcnow().isoformat()}
|
|
write_checkpoint(root_path, snapshot)
|
|
log("Index.Rebuild", "REFLECT", "Rebuild complete", {"contracts": len(contracts)})
|
|
return snapshot
|
|
# #endregion Index.Rebuild
|
|
```
|
|
|
|
## 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 Api.Dashboards [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 Dashboards.List [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 Dashboards.List
|
|
|
|
# #endregion Api.Dashboards
|
|
```
|
|
|
|
### SQLAlchemy model pattern
|
|
```python
|
|
# #region Models.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 Models.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
|