busya
9b8c485562
- Replace pylint with ruff (backend/ruff.toml) - line-length=300, max-complexity=10, per-file-ignores for tests - Exclude D (docstrings replaced by contracts) and ANN - Add eslint flat config for frontend (eslint-plugin-svelte) - Allow console.info/warn/error (belief-state protocol) - Remove .pylintrc (references non-existent plugin) - Update agent docs: ruff check . + npm run lint - Add ruff>=0.11.0 to backend/requirements.txt - Add eslint + plugins to frontend devDependencies
283 lines
11 KiB
Markdown
283 lines
11 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,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]
|
|
@RELATION DEPENDS_ON -> [MolecularCoTLogging]
|
|
|
|
## 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 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
|
|
```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
|