Modules · wrap-up
Cheat sheet & self-check
Every command from this section on one page.
Environment
python3 -m venv .venv | Create a virtual environment in .venv |
source .venv/bin/activate | Use it in this shell (deactivate to leave) |
pip install requests pyyaml | Install packages into the venv |
pip freeze > requirements.txt | Record exact versions |
python3 script.py | Run a script |
Core syntax
hosts = ["web-01", "web-02"] | A list |
ports = {"http": 80, "https": 443} | A dictionary |
for h in hosts: print(h) | Loop |
f"{host} has {len(disks)} disks" | f-string formatting |
try: … except FileNotFoundError: … | Handle a specific error |
Files
Path('app.conf').read_text() | Whole file as a string |
for line in Path('hosts.txt').read_text().splitlines(): | Line by line |
Path('out.txt').write_text(data) | Write a file |
list(Path('/var/log').glob('*.log')) | Find files |
Formats
json.loads(text) / json.dumps(obj, indent=2) | JSON string ↔ Python |
yaml.safe_load(text) | YAML → Python (safe: never yaml.load) |
yaml.safe_dump(obj, sort_keys=False) | Python → YAML |
csv.DictReader(f) / csv.DictWriter(f, fieldnames=…) | CSV rows as dicts |
subprocess.run
subprocess.run(["systemctl", "is-active", "nginx"]) | Run a command (arguments as a list) |
…, check=True | Raise CalledProcessError on non-zero exit |
…, capture_output=True, text=True | Capture stdout/stderr as strings |
…, timeout=30 | Kill it and raise TimeoutExpired after 30 s |
result.returncode, result.stdout, result.stderr | What happened |
Safety
shell=False (the default) | No shell: no injection, no globbing surprises |
shlex.split('ls -l "/my dir"') | Split a command string like a shell would |
shlex.quote(user_input) | If you truly need a shell, quote untrusted values |
requests basics
r = requests.get(url, params={'state': 'open'}, timeout=10) | GET with query parameters and a timeout |
r.raise_for_status() | Raise for 4xx/5xx responses |
r.json() | Parse the JSON body |
requests.post(url, json=payload, timeout=10) | POST a JSON body |
s = requests.Session(); s.headers['Authorization'] = f'Bearer {token}' | Reuse connections and headers |
HTTP status classes
2xx | Success |
4xx | Your request is wrong (401 auth, 403 forbidden, 404 missing, 429 rate limited) |
5xx | Server-side problem: often worth retrying |
argparse
p = argparse.ArgumentParser(description=…) | Create the parser |
p.add_argument('--dry-run', action='store_true') | A boolean flag |
p.add_argument('-n', '--namespace', default='default') | An option with a default |
sub = p.add_subparsers(dest='cmd', required=True) | Subcommands (tool list / tool drain …) |
logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') | Configure once, at start-up |
log = logging.getLogger(__name__) | A logger per module |
log.info('drained %s', node) | Lazy formatting: args, not f-strings |
-v / -q flags → DEBUG / WARNING | Let users choose verbosity |
Connect
pip install kubernetes | The official client |
config.load_kube_config() | Use ~/.kube/config (current context) |
config.load_incluster_config() | Inside a pod: use its ServiceAccount |
v1 = client.CoreV1Api(); apps = client.AppsV1Api() | API groups: core (pods, nodes…) and apps (deployments…) |
Common calls
v1.list_pod_for_all_namespaces(label_selector='app=web') | List pods with a label |
v1.list_node() | List nodes |
apps.patch_namespaced_deployment_scale(name, ns, {'spec': {'replicas': 3}}) | Scale a Deployment |
watch.Watch().stream(v1.list_namespaced_pod, 'default', timeout_seconds=60) | Stream changes |
Setup
pip install boto3 | The AWS SDK for Python |
boto3.Session(profile_name='dev', region_name='eu-west-1') | Explicit profile and region |
session.client('ec2') | A low-level client for a service |
session.client('sts').get_caller_identity() | Who am I? (account and ARN) |
Patterns
ec2.get_paginator('describe_instances').paginate(Filters=[…]) | All pages, not just the first |
ec2.get_waiter('instance_running').wait(InstanceIds=[…]) | Wait until a state is reached |
except botocore.exceptions.ClientError as e: e.response['Error']['Code'] | Handle specific AWS errors |
pytest
pip install pytest | Install |
pytest -q | Run all tests quietly |
pytest -k disk -x | Only tests matching 'disk', stop at first failure |
@pytest.mark.parametrize('pct,expected', [(50, 'OK'), (95, 'CRITICAL')]) | One test, many cases |
def test_x(tmp_path): … | A temporary directory per test |
Quality & packaging
ruff check . && ruff format . | Lint and format |
pip install -e . | Install your project in editable mode |
[project.scripts] opsctl = "opsctl.cli:main" | Make 'opsctl' a real command |
python -m build | Build a wheel to distribute |
fleetreport usage (what you'll build)
fleetreport | Report on every context in the kubeconfig |
fleetreport --contexts prod-eu,prod-us -o json | Selected clusters, JSON output |
fleetreport --workers 8 --timeout 20 | Parallelism and per-cluster timeout |
echo $? | 0 = all healthy, 1 = findings, 2 = usage, 3 = could not reach a cluster |
