Code Style¶
Style guide and conventions for Machineuse development.
Overview¶
We use automated tools to enforce consistent style:
| Tool | Purpose |
|---|---|
| Black | Code formatting |
| isort | Import sorting |
| Flake8 | Linting |
| mypy | Type checking |
Running Style Checks¶
# Format code
poetry run black machineuse tests
poetry run isort machineuse tests
# Lint
poetry run flake8 machineuse tests
# Type check
poetry run mypy machineuse
Python Style¶
Formatting (Black)¶
Black handles formatting automatically. Key settings:
Imports (isort)¶
Imports are organized in groups:
# Standard library
import os
import sys
from datetime import datetime
from typing import Dict, List, Optional
# Third-party
import asyncio
import pytest
from fastapi import FastAPI
# Local
from machineuse.core.config import Config
from machineuse.core.storage import StorageProvider
Configuration:
Linting (Flake8)¶
# .flake8
[flake8]
max-line-length = 88
extend-ignore = E203, E501
exclude = .git,__pycache__,build,dist
Type Hints¶
Required for Public APIs¶
# Good
def create_instance(
image: str,
config: Optional[Dict[str, Any]] = None
) -> Instance:
"""Create a new container instance."""
...
# Avoid
def create_instance(image, config=None):
...
Type Hint Patterns¶
from typing import Dict, List, Optional, Union, TypeVar, Generic
# Optional values
def get_instance(id: str) -> Optional[Instance]:
...
# Collections
def list_instances(filters: Dict[str, str]) -> List[Instance]:
...
# Callbacks
from typing import Callable
def on_event(callback: Callable[[Event], None]) -> None:
...
# Generics
T = TypeVar('T')
class Repository(Generic[T]):
def get(self, id: str) -> Optional[T]:
...
mypy Configuration¶
# pyproject.toml
[tool.mypy]
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
disallow_untyped_defs = true
[[tool.mypy.overrides]]
module = "tests.*"
disallow_untyped_defs = false
Naming Conventions¶
Variables and Functions¶
# Variables: snake_case
instance_id = "abc123"
max_instances = 50
# Functions: snake_case
def create_instance():
...
async def get_cluster_status():
...
Classes¶
# Classes: PascalCase
class InstanceManager:
...
class StorageProvider:
...
# Exceptions
class InstanceNotFoundError(Exception):
...
Constants¶
Private Members¶
class Storage:
def __init__(self):
self._connection = None # Protected
self.__cache = {} # Private (name mangling)
def _internal_method(self): # Protected method
...
Documentation¶
Docstrings (Google Style)¶
def create_instance(
image: str,
config: Optional[Dict[str, Any]] = None,
node_preference: Optional[str] = None
) -> Instance:
"""Create a new container instance.
Creates an isolated browser container with the specified image
and configuration. The scheduler will place the instance on
an appropriate worker node.
Args:
image: Base container image (e.g., "ubuntu:22.04").
config: Optional instance configuration including memory,
CPU, and disk settings.
node_preference: Optional preferred worker node ID.
Returns:
The created Instance object with assigned ID and status.
Raises:
InsufficientResourcesError: If no node has available capacity.
InvalidImageError: If the specified image is not available.
Example:
>>> instance = create_instance(
... image="ubuntu:22.04",
... config={"memory_gb": 4}
... )
>>> print(instance.id)
'abc123def456'
"""
...
Module Docstrings¶
"""Container management module.
This module provides the core functionality for managing
systemd-nspawn containers, including creation, lifecycle
management, and resource monitoring.
Example:
>>> from machineuse.core.containers import ContainerManager
>>> manager = ContainerManager()
>>> container = await manager.create("test-1")
"""
Error Handling¶
Custom Exceptions¶
class MachineUseError(Exception):
"""Base exception for Machineuse errors."""
pass
class InstanceNotFoundError(MachineUseError):
"""Raised when an instance is not found."""
def __init__(self, instance_id: str):
self.instance_id = instance_id
super().__init__(f"Instance not found: {instance_id}")
class InsufficientResourcesError(MachineUseError):
"""Raised when resources are insufficient."""
pass
Exception Handling¶
# Good: Specific exceptions
try:
instance = await storage.get_instance(instance_id)
except InstanceNotFoundError:
raise HTTPException(status_code=404, detail="Instance not found")
except StorageError as e:
logger.error(f"Storage error: {e}")
raise HTTPException(status_code=500, detail="Internal error")
# Avoid: Bare except
try:
...
except: # Bad!
pass
Async Patterns¶
Async Functions¶
# Async function
async def get_instance(instance_id: str) -> Instance:
async with storage.connection() as conn:
return await conn.fetch_one(instance_id)
# Async context manager
class DatabaseConnection:
async def __aenter__(self):
self._conn = await aiosqlite.connect(self.path)
return self._conn
async def __aexit__(self, *args):
await self._conn.close()
Concurrent Operations¶
# Run tasks concurrently
async def get_all_node_status():
tasks = [get_node_status(node) for node in nodes]
return await asyncio.gather(*tasks)
# With error handling
results = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
logger.error(f"Task failed: {result}")
File Organization¶
Module Structure¶
machineuse/
├── __init__.py # Package exports
├── core/
│ ├── __init__.py
│ ├── config.py # Configuration
│ ├── storage/
│ │ ├── __init__.py
│ │ ├── base.py # Abstract interface
│ │ ├── sqlite.py # SQLite implementation
│ │ └── postgres.py # PostgreSQL implementation
│ └── containers/
│ ├── __init__.py
│ ├── runtime.py # Container runtime
│ └── snapshot.py # Snapshot management
├── api/
│ ├── __init__.py
│ ├── server.py # FastAPI app
│ └── routes/
│ ├── instances.py
│ └── nodes.py
└── nodes/
├── agent.py
└── control_plane.py