Writing Python Code
Prerequisites
This skill extends myai's engineering-principles. Load that first. using-my-skills and engineering-principles are assumed already loaded via myai bootstrap.
For the general engineering philosophy (pit of success, fail fast, error handling as control flow, testing philosophy, architecture separation), see myai's engineering-principles. This skill covers only Python-specific tooling, type system configuration, and patterns.
Run project-local tools through uv, not system-installed binaries: uv run python, uv run pytest, uv run ruff, uv run basedpyright, uv run poe, uv run pre-commit. The normal verification pair is uv run poe lint_full and uv run poe test.
Type System
basedpyright Configuration
[tool.pyright]
pythonVersion = "3.14"
typeCheckingMode = "strict"
reportAny = "error"
reportImportCycles = "error"
reportImplicitStringConcatenation = "none"
reportUnusedCallResult = "none"
reportUnnecessaryIsInstance = "none"
Additional strict options (for medium-big projects):
reportExplicitAny = "error"
reportUnnecessaryTypeIgnoreComment = "error"
reportMissingModuleSource = "error"
reportPrivateUsage = "error"
reportOptionalMemberAccess = "error"
reportOptionalCall = "error"
reportAttributeAccessIssue = "error"
Banned Patterns
| Banned | Use Instead |
|---|---|
Any |
Banned entirely — define the actual type |
object (unrestricted) |
Restricted to boundary positions only (see below) |
typing.cast() |
isinstance, TypeIs, pattern matching |
# type: ignore without rationale |
# type: ignore[specific-code] # rationale: <reason> |
Raw dict in business logic |
msgspec.Struct, dataclass, or TypedDict |
| Implicit return types | Explicit annotation on every function |
object is allowed only in: TypeIs/TypeGuard params, *args: object, Coroutine[object, None, T], PySide6 Signal(object). Everywhere else, use Protocol, TypeVar, union types, or typed structures.
Type Patterns
External data → msgspec.Struct:
import msgspec
class UserConfig(msgspec.Struct):
name: str
port: int
debug: bool = False
# JSON → typed object (validates at decode time)
config = msgspec.json.decode(raw_bytes, type=UserConfig)
# Dict/YAML → typed object
config = msgspec.convert(raw_dict, type=UserConfig)
TypedDict is still valid when dict compatibility is needed (e.g., **unpacking, APIs expecting dicts).
FastAPI HTTP edge → pydantic DTOs are used at the transport boundary. Convert them immediately into framework-free typed structures and do not let framework models become domain models.
Domain objects → dataclass:
@dataclass(frozen=True, slots=True)
class Profile:
name: str
version: str
active: bool = True
Duck typing → Protocol:
class Renderable(Protocol):
def render(self) -> str: ...
Constants → Final:
MAX_RETRIES: Final = 3
CONFIG_PATH: Final[Path] = Path("~/.config/app").expanduser()
Handling Any at Library Boundaries
0. Typed deserialization (for external data):
msgspec.json.decode(data, type=MyStruct) eliminates Any for JSON/API responses — the return type is MyStruct, not Any. Use this before reaching for wrappers when the boundary is data deserialization.
1. Typed wrappers (preferred for library APIs):
class WhisperModelWrapper:
def __init__(self, model_size: str, device: str = "auto") -> None:
from faster_whisper import WhisperModel as _WhisperModel
self._model = _WhisperModel(model_size, device=device)
def transcribe(self, audio: np.ndarray, language: str | None = None) -> TranscriptionResult:
segments_gen, info = self._model.transcribe(audio, language=language)
return TranscriptionResult(
text="".join(s.text for s in segments_gen),
language=str(info.language),
)
Enforce wrapper usage via ruff:
[tool.ruff.lint.flake8-tidy-imports.banned-api]
"faster_whisper" = { msg = "Use src/wrappers/whisper_wrapper instead" }
2. Type stubs in src/stubs/:
# src/stubs/some_library.pyi
def some_function(arg: str) -> list[int]: ...
Configure: stubPath = "src/stubs" in basedpyright config.
3. Inline type narrowing for one-off cases:
raw_value = untyped_lib.get_value() # returns Any
assert isinstance(raw_value, str) # narrows to str
TypeIs Guards
Note:
TypeIsguards are unnecessary formsgspec-decoded data (already fully typed). Use them for narrowing in-memory objects of unknown type.
from typing import TypeIs, TypedDict, Required
class ValidResponse(TypedDict):
status: Required[str]
data: Required[dict[str, str | int | bool | list[str]]]
# Simplified for brevity — production code should use msgspec.convert() for full validation
def is_valid_response(obj: object) -> TypeIs[ValidResponse]:
return (
isinstance(obj, dict)
and isinstance(obj.get("status"), str)
and isinstance(obj.get("data"), dict)
)
def process(response: object) -> Result[str, str]:
if not is_valid_response(response):
return Err("Invalid response format")
return Ok(response["data"]["key"]) # type-safe access
Pattern Matching for Type Safety
@dataclass
class Success:
value: str
@dataclass
class Failure:
error: str
code: int
type Outcome = Success | Failure
def handle(outcome: Outcome) -> str:
match outcome:
case Success(value=v):
return f"OK: {v}"
case Failure(error=e, code=c):
return f"Error {c}: {e}"
# type: ignore Policy
Every # type: ignore must have a specific error code and rationale:
# BAD
result = some_call() # type: ignore
# GOOD
result = some_call() # type: ignore[no-any-return] # rationale: lib returns Any, validated below
assert isinstance(result, ExpectedType)
TYPE_CHECKING Guard
For imports that cause circular dependencies or are only needed for annotations:
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from src.managers.audio_manager import AudioManager
class TranscriptionService:
def __init__(self, audio: "AudioManager") -> None: ...
Note:
TYPE_CHECKINGis for forward references within the same layer. If two modules import each other, that's a circular dependency — fix the architecture (extract a Protocol, move types to a common module), don't hide the cycle withTYPE_CHECKING.
Error Handling
For the general principle of error handling as control flow, see myai's
engineering-principles. Python-specific: useResult[T, E]fromrusty-resultsfor expected failures. Exceptions are reserved for programming errors (bugs).
Decision Table
| Situation | Use |
|---|---|
| File not found, network error, invalid input | Result[T, E] |
| User provided bad data | Result[T, E] |
| Third-party library raised | Catch at boundary → Result[T, E] |
| Invariant violated (should never happen) | raise exception |
| Invalid program state (bug) | raise exception |
Pattern
import msgspec
from rusty_results import Result, Ok, Err
class Config(msgspec.Struct):
name: str
def load_config(path: Path) -> Result[Config, str]:
if not path.exists():
return Err(f"Config not found: {path}")
try:
return Ok(msgspec.json.decode(path.read_bytes(), type=Config))
except (msgspec.DecodeError, OSError) as e:
return Err(f"Failed to load: {e}")
Rules
Early returns — handle error first, keep success path linear:
result = load_data(path) if result.is_err: return Err(f"Cannot proceed: {result.unwrap_err()}") data = result.unwrap()Three error boundaries:
- Library: catch third-party exceptions → Result
- Component: each subsystem returns Result to caller
- Global: UI/CLI top-level catches everything, shows user message
Never swallow errors — no
except: pass, no ignored Results. EveryResult[T, E]must be checked — at minimum log the error + show toast/alert in GUI or print to CLI.Cleanup on failure — if multi-step operation fails midway, clean up partial state
Custom error types for complex domains:
@dataclass class ConfigError: path: Path reason: str line: int | None = NoneHandle received error values gracefully: show UI warning, or log, or do early return or propagate with context
Async Task Boundaries
Any coroutine launched via asyncio.ensure_future() or create_task() is a fire-and-forget boundary. If an exception escapes, nobody retrieves it and the UI gets stuck in an intermediate state (e.g. "processing" forever).
Mandatory pattern: wrap the entire coroutine body in try/except Exception as a safety net:
async def _do_work(self, path: Path) -> None:
try:
await self._do_work_inner(path)
except Exception as exc:
logger.exception("Unexpected error for %s", path)
self._set_error_state(path, f"Unexpected error: {exc}")
The inner method handles expected errors (Result checks, specific exceptions). The outer method guarantees the UI always transitions to a terminal state.
CLI Error Boundary
def main() -> int:
result = run_app()
if result.is_err:
typer.echo(f"Error: {result.unwrap_err()}", err=True)
return 1
return 0
Async Patterns
When to Use
| Use async | Don't use async |
|---|---|
| File I/O | Pure data transformations |
| Network requests | Simple calculations |
| Subprocess execution | In-memory operations |
| Operations >100ms | Quick lookups |
| Parallel I/O operations | Sequential pure logic |
HTTP Requests
async def fetch_data(url: str) -> Result[bytes, str]:
try:
async with httpx.AsyncClient() as client:
resp = await client.get(url, timeout=30.0)
resp.raise_for_status()
return Ok(resp.content)
except httpx.HTTPError as e:
return Err(f"HTTP error: {e}")
Subprocess Execution
async def run_command(args: list[str]) -> Result[str, str]:
try:
process = await asyncio.create_subprocess_exec(
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
return Err(f"Command failed: {stderr.decode()}")
return Ok(stdout.decode())
except FileNotFoundError:
return Err(f"Command not found: {args[0]}")
Concurrent Operations
async def fetch_all(urls: list[str]) -> list[Result[bytes, str]]:
return await asyncio.gather(*(fetch_data(url) for url in urls))
Timeouts
async def fetch_with_timeout(url: str) -> Result[bytes, str]:
try:
async with asyncio.timeout(10):
return await fetch_data(url)
except TimeoutError:
return Err(f"Timeout fetching {url}")
Async Rules
- Never call
subprocess.run(),time.sleep(), or synchronous HTTP in async context - Never use
shell=Truein subprocess calls - Always use
asyncio.create_subprocess_exec()for subprocesses - Always use
async withfor resource management (HTTP clients, file handles) - Never mix sync and async in the same call chain
- Always handle
TimeoutErrorfor operations that may hang
Code Style
| Rule | Value |
|---|---|
| Line length | 120 |
| Indentation | 4 spaces |
| Quotes | Double quotes (single only to avoid escaping) |
| Import order | stdlib → third-party → local (auto-sorted by ruff) |
Naming
| Element | Convention | Example |
|---|---|---|
| Modules, functions, variables | snake_case |
load_config, user_name |
| Classes, TypedDicts | PascalCase |
ProfileManager, UserConfig |
| Constants | UPPER_SNAKE |
MAX_RETRIES, DEFAULT_PORT |
| Private | _prefix |
_internal_state |
Documentation
Google-style docstrings on public APIs. Comments explain why, not what.
Preconditions & Validation
For the general fail-fast principle, see myai's
engineering-principles. Python-specific: validate at subsystem entry points usingmsgspec.Structfor data shape,isinstance/TypeIsfor type narrowing, andResult[T,E]for expected validation failures. Check permissions, external deps, config validity, and input ranges before proceeding with business logic.
Architecture
For the general architecture separation principle, see myai's
engineering-principlesandarchitecting-changes. Python-specific layout:
Presentation (Qt GUI / CLI / API)
|
v
Domain (Managers, Models, Business Rules)
|
v
Utilities (Helpers, Wrappers, Common)
- Dependencies flow downward only
- UI is a plugin — adding CLI, GUI, or API should not change business logic
- Domain never imports from presentation
- For architecture decisions, load
architecting-python-changes(which extends myai'sarchitecting-changes)
Security
- No
shell=Truein subprocess calls - Validate paths (symlink-aware):
def is_safe_path(path: Path, base: Path) -> bool: try: resolved = path.resolve(strict=True) return resolved.is_relative_to(base.resolve(strict=True)) except (OSError, ValueError): return False - No hardcoded secrets — use environment variables or dotenv
- Input validation at system boundaries
- Never interpolate user input into subprocess commands
- Cleanup on failure — if multi-step operation fails midway, clean up partial state
Performance
- Clear code first, optimize only with profiling data
__slots__on frequently instantiated dataclasses- Batch I/O operations (don't read files one by one in a loop)
- Lazy loading for expensive resources (load on first use, not import time)
- Concurrent I/O with
asyncio.gather()
Logging
import colorlog
def get_logger(name: str) -> logging.Logger:
handler = colorlog.StreamHandler()
handler.setFormatter(colorlog.ColoredFormatter(
"%(log_color)s%(levelname)-8s%(reset)s %(name)s: %(message)s"
))
logger = logging.getLogger(name)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return logger
Usage: logger = get_logger(__name__)
Jinja2 Templating
Use when generating text output (HTML, configs, reports, markdown):
Tooling
| Tool | Purpose |
|---|---|
uv |
Package management, script execution |
basedpyright |
Type checking (strict) |
ruff |
Strict lint + format |
pytest |
Testing |
rusty-results |
Result[T, E] pattern |
typer |
CLI framework (preferred) — always configure -h support (see below) |
argparse |
CLI only for stdlib-only scripts |
PySide6 |
GUI (no system deps) |
httpx |
HTTP (async) |
msgspec |
External data validation + parsing |
Jinja2 |
Text output generation |
Always enable -h for help — typer only supports --help by default:
app = typer.Typer(context_settings={"help_option_names": ["-h", "--help"]})
Run uv run poe lint_full continuously, not just at the end. Treat it as the main sweep for basedpyright, Ruff check/format, and custom linters; pair it with uv run poe test for verification.
Related myai Skills
engineering-principles— Parent skill. Language-agnostic philosophy: pit of success, fail fast, error handling as control flow, testing philosophy, architecture separation.architecting-changes— For architecture decisions before writing code. Loadarchitecting-python-changesfor the Python-specific extension.code-simplification— For refactoring Python code for clarity.