|
| 1 | +# Spacecraft Python Guidelines — Full Reference |
| 2 | + |
| 3 | +**Version:** 1.0 |
| 4 | +**Date:** 2026-07-13 |
| 5 | +**Author:** Mohamed Hammad & Spacecraft Software |
| 6 | +**Compatibility:** Claude 3.5+, Claude 4, Grok, and all advanced reasoning models |
| 7 | + |
| 8 | +This document expands on the `SKILL.md` for Python systems programming. It provides complete, compile-checked configurations and skeletons for ProcessPoolExecutor parallelism, asyncio execution, Pydantic v2 validations, slots optimizations, and testing. |
| 9 | + |
| 10 | +--- |
| 11 | + |
| 12 | +## 1. Concurrency: ProcessPoolExecutor & Asyncio run_in_executor |
| 13 | + |
| 14 | +Do not block the single-threaded `asyncio` event loop. Offload CPU-heavy calculations to a process pool (bypassing the GIL) and synchronous blocking operations to thread executors. |
| 15 | + |
| 16 | +```python |
| 17 | +import asyncio |
| 18 | +from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor |
| 19 | +import time |
| 20 | +from typing import List |
| 21 | + |
| 22 | +# Helper CPU-bound math calculation |
| 23 | +def compute_factorial_sync(n: int) -> int: |
| 24 | + if n < 0: |
| 25 | + raise ValueError("Must be non-negative") |
| 26 | + res = 1 |
| 27 | + for i in range(2, n + 1): |
| 28 | + res *= i |
| 29 | + return res |
| 30 | + |
| 31 | +# Helper blocking I/O operation |
| 32 | +def blocking_file_read(filepath: str) -> str: |
| 33 | + # Simulates blocking OS operation |
| 34 | + time.sleep(0.1) |
| 35 | + with open(filepath, "r", encoding="utf-8") as f: |
| 36 | + return f.read() |
| 37 | + |
| 38 | +class TelemetryManager: |
| 39 | + def __init__(self, max_processes: int = 4, max_threads: int = 8) -> None: |
| 40 | + self.process_pool = ProcessPoolExecutor(max_workers=max_processes) |
| 41 | + self.thread_pool = ThreadPoolExecutor(max_workers=max_threads) |
| 42 | + |
| 43 | + async def shutdown(self) -> None: |
| 44 | + self.process_pool.shutdown(wait=True) |
| 45 | + self.thread_pool.shutdown(wait=True) |
| 46 | + |
| 47 | + async def calculate_batch_factorials(self, numbers: List[int]) -> List[int]: |
| 48 | + """Runs GIL-bound CPU math inside a ProcessPoolExecutor concurrently.""" |
| 49 | + loop = asyncio.get_running_loop() |
| 50 | + tasks = [ |
| 51 | + loop.run_in_executor(self.process_pool, compute_factorial_sync, num) |
| 52 | + for num in numbers |
| 53 | + ] |
| 54 | + results: List[int] = await asyncio.gather(*tasks) |
| 55 | + return results |
| 56 | + |
| 57 | + async def read_files_async(self, filepaths: List[str]) -> List[str]: |
| 58 | + """Runs blocking system operations inside ThreadPoolExecutor to prevent event loop lag.""" |
| 59 | + loop = asyncio.get_running_loop() |
| 60 | + tasks = [ |
| 61 | + loop.run_in_executor(self.thread_pool, blocking_file_read, path) |
| 62 | + for path in filepaths |
| 63 | + ] |
| 64 | + results: List[str] = await asyncio.gather(*tasks) |
| 65 | + return results |
| 66 | +``` |
| 67 | + |
| 68 | +--- |
| 69 | + |
| 70 | +## 2. Typing & Pydantic v2 Ingress Validation |
| 71 | + |
| 72 | +Validate and parse raw dynamic data immediately at the entrance boundaries using Pydantic (v2) models. Use strict typing annotations for mypy. |
| 73 | + |
| 74 | +```python |
| 75 | +from datetime import datetime |
| 76 | +import json |
| 77 | +from uuid import UUID |
| 78 | +from pydantic import BaseModel, ConfigDict, Field, ValidationError |
| 79 | +from typing import Union, Dict, Any |
| 80 | + |
| 81 | +class TelemetryPacket(BaseModel): |
| 82 | + # Configure model to be immutable and forbid extra parameters |
| 83 | + model_config = ConfigDict( |
| 84 | + frozen=True, |
| 85 | + extra="forbid" |
| 86 | + ) |
| 87 | + |
| 88 | + packet_id: UUID = Field(alias="packetId") |
| 89 | + value: float = Field(ge=0.0) # validation: must be >= 0.0 |
| 90 | + timestamp: datetime |
| 91 | + |
| 92 | +# Functional result mappings |
| 93 | +class ValidationSuccess: |
| 94 | + __slots__ = ("data",) |
| 95 | + def __init__(self, data: TelemetryPacket) -> None: |
| 96 | + self.data = data |
| 97 | + |
| 98 | +class ValidationFailure: |
| 99 | + __slots__ = ("error",) |
| 100 | + def __init__(self, error: str) -> None: |
| 101 | + self.error = error |
| 102 | + |
| 103 | +ValidationResult = Union[ValidationSuccess, ValidationFailure] |
| 104 | + |
| 105 | +def parse_incoming_payload(json_str: str) -> ValidationResult: |
| 106 | + try: |
| 107 | + # Pydantic v2 direct JSON parsing (faster than json.loads) |
| 108 | + packet = TelemetryPacket.model_validate_json(json_str) |
| 109 | + return ValidationSuccess(packet) |
| 110 | + except ValidationError as e: |
| 111 | + return ValidationFailure(str(e)) |
| 112 | +``` |
| 113 | + |
| 114 | +--- |
| 115 | + |
| 116 | +## 3. Slots Attribute Memory Optimization |
| 117 | + |
| 118 | +For classes that hold data and are instantiated frequently (e.g. millions of packet structures), define `__slots__` to prevent dynamic `__dict__` generation, lowering memory usage. |
| 119 | + |
| 120 | +```python |
| 121 | +from uuid import UUID |
| 122 | +from typing import Tuple |
| 123 | + |
| 124 | +class CompactReading: |
| 125 | + # Explicit slots reduce memory layout size and speed up variable lookup |
| 126 | + __slots__ = ("sensor_id", "measurements") |
| 127 | + |
| 128 | + def __init__(self, sensor_id: UUID, measurements: Tuple[float, ...]) -> None: |
| 129 | + self.sensor_id: UUID = sensor_id |
| 130 | + self.measurements: Tuple[float, ...] = measurements |
| 131 | + |
| 132 | + def average(self) -> float: |
| 133 | + if not self.measurements: |
| 134 | + return 0.0 |
| 135 | + return sum(self.measurements) / len(self.measurements) |
| 136 | +``` |
| 137 | + |
| 138 | +--- |
| 139 | + |
| 140 | +## 4. Testing: pytest & pytest-asyncio |
| 141 | + |
| 142 | +Test concurrent routines and validations using `pytest` and `pytest-asyncio`. Ensure warnings are treated as errors. |
| 143 | + |
| 144 | +```python |
| 145 | +# tests/test_telemetry.py |
| 146 | +import pytest |
| 147 | +from uuid import uuid4 |
| 148 | +from datetime import datetime, timezone |
| 149 | +from telemetry_manager import TelemetryManager |
| 150 | +from telemetry_schemas import parse_incoming_payload, ValidationSuccess, ValidationFailure |
| 151 | + |
| 152 | +@pytest.fixture |
| 153 | +def manager() -> TelemetryManager: |
| 154 | + return TelemetryManager(max_processes=2, max_threads=2) |
| 155 | + |
| 156 | +@pytest.mark.asyncio |
| 157 | +async def test_factorial_calculations(manager: TelemetryManager) -> None: |
| 158 | + numbers = [5, 6, 7] |
| 159 | + results = await manager.calculate_batch_factorials(numbers) |
| 160 | + assert results == [120, 720, 5040] |
| 161 | + |
| 162 | +def test_pydantic_valid_parsing() -> None: |
| 163 | + uuid_str = str(uuid4()) |
| 164 | + payload = f'{{"packetId": "{uuid_str}", "value": 45.2, "timestamp": "2026-07-12T20:00:00Z"}}' |
| 165 | + |
| 166 | + result = parse_incoming_payload(payload) |
| 167 | + |
| 168 | + assert isinstance(result, ValidationSuccess) |
| 169 | + assert result.data.value == 45.2 |
| 170 | + assert result.data.timestamp.tzinfo == timezone.utc |
| 171 | + |
| 172 | +def test_pydantic_invalid_parsing() -> None: |
| 173 | + # Value violates constraint (must be >= 0.0) |
| 174 | + payload = '{"packetId": "invalid-uuid", "value": -10.0, "timestamp": "2026-07-12T20:00:00Z"}' |
| 175 | + |
| 176 | + result = parse_incoming_payload(payload) |
| 177 | + |
| 178 | + assert isinstance(result, ValidationFailure) |
| 179 | +``` |
| 180 | + |
| 181 | +--- |
| 182 | + |
| 183 | +## 5. Tooling Configuration (`pyproject.toml`) |
| 184 | + |
| 185 | +Standard configuration mapping for `Ruff` rules, `Mypy` strict settings, and `Pytest` warnings. |
| 186 | + |
| 187 | +```toml |
| 188 | +[tool.poetry] |
| 189 | +name = "telemetry-service" |
| 190 | +version = "0.1.0" |
| 191 | +description = "Spacecraft Telemetry Service in Python" |
| 192 | +authors = ["Mohamed Hammad <Mohamed.Hammad@SpacecraftSoftware.org>"] |
| 193 | +license = "GPL-3.0-or-later" |
| 194 | + |
| 195 | +[tool.poetry.dependencies] |
| 196 | +python = "^3.12" |
| 197 | +pydantic = "^2.5.0" |
| 198 | +pytest = "^7.4.0" |
| 199 | +pytest-asyncio = "^0.21.0" |
| 200 | +ruff = "^0.1.0" |
| 201 | +mypy = "^1.6.0" |
| 202 | + |
| 203 | +[tool.mypy] |
| 204 | +strict = true |
| 205 | +disallow_any_generics = true |
| 206 | +warn_unused_ignores = true |
| 207 | +plugins = ["pydantic.mypy"] |
| 208 | + |
| 209 | +[tool.pydantic-mypy] |
| 210 | +init_forbid_extra = true |
| 211 | +init_typed = true |
| 212 | +warn_required_dynamic_aliases = true |
| 213 | + |
| 214 | +[tool.ruff] |
| 215 | +target-version = "py312" |
| 216 | +line-length = 100 |
| 217 | + |
| 218 | +[tool.ruff.lint] |
| 219 | +select = [ |
| 220 | + "E", "F", "W", # Pycodestyle & Pyflakes |
| 221 | + "B", # Flake8-bugbear |
| 222 | + "I", # Isort |
| 223 | + "RUF", # Ruff-specific rules |
| 224 | + "ANN", # Type annotations |
| 225 | + "ASYNC", # Asyncio lints |
| 226 | + "TCH", # Type checking blocks |
| 227 | + "UP" # Pyupgrade |
| 228 | +] |
| 229 | + |
| 230 | +[tool.ruff.format] |
| 231 | +quote-style = "double" |
| 232 | +indent-style = "space" |
| 233 | + |
| 234 | +[tool.pytest.ini_options] |
| 235 | +minversion = "7.0" |
| 236 | +addopts = "-ra -q" |
| 237 | +testpaths = ["tests"] |
| 238 | +filterwarnings = [ |
| 239 | + "error", # Treat all warnings as compilation errors |
| 240 | +] |
| 241 | +``` |
| 242 | + |
| 243 | +--- |
| 244 | + |
| 245 | +## 6. Common Pitfalls & Troubleshooting |
| 246 | + |
| 247 | +| Pitfall | Symptom | Corrective Action | |
| 248 | +| :--- | :--- | :--- | |
| 249 | +| **Synchronous sleep in async** | Event loop blocks, other jobs halt | Replace `time.sleep` with `await asyncio.sleep`. | |
| 250 | +| **Synchronous filesystem in async** | Event loop thread freezes | Offload execution using `loop.run_in_executor(None, sync_func)`. | |
| 251 | +| **Threading for math operations** | High CPU context switches, slow speed | Replace threading with `ProcessPoolExecutor`. | |
| 252 | +| **Loose typing annotations** | Mypy fails in strict mode | Annotate variables and parameters explicitly. | |
| 253 | +| **Dynamic properties on dataclasses**| Memory growth on long processes | Declare class variables inside `__slots__` tuple. | |
| 254 | +| **Mocking Pydantic dynamic fields** | Mypy failures on class initializers | Configure Pydantic Mypy plugin with `init_typed = true`. | |
| 255 | + |
| 256 | +--- |
| 257 | + |
| 258 | +## 7. Code Review Compliance Gate |
| 259 | + |
| 260 | +Before merging Python code, verify: |
| 261 | +1. Static typing assertions pass under strict Mypy checks (no untyped parameters). |
| 262 | +2. Asynchronous loops do not execute synchronous blocking operations. |
| 263 | +3. Heavy numerical calculations run in separate processes (`ProcessPoolExecutor`). |
| 264 | +4. High frequency data structures define `__slots__` tuples. |
| 265 | +5. Ingress parsing checks are wrapped in Pydantic models configured with `frozen=True`. |
| 266 | +6. Ruff lints pass cleanly and pytest treats warnings as error configurations. |
0 commit comments