Python cho DevOps — Automation & Infrastructure Scripting

Python là ngôn ngữ scripting số 1 trong DevOps. Ansible, AWS CDK, Terraform modules, SaltStack đều dùng Python. Nếu bạn có thể viết Python cơ bản, bạn có thể tự động hóa gần như mọi tác vụ vận hành.

---

Tại sao Python, không phải Bash?

Rule of thumb: Nếu script dài > 50 dòng và có nhiều điều kiện, logic → dùng Python.

---

Cài đặt và Môi trường

# Kiểm tra Python có sẵn
python3 --version
pip3 --version

# Ubuntu/Debian
sudo apt install python3 python3-pip python3-venv

# RHEL/Rocky
sudo dnf install python3 python3-pip

# Tạo virtual environment (cô lập dependencies)
python3 -m venv venv
source venv/bin/activate           # Linux/macOS
deactivate                         # thoát venv

# Cài packages vào venv
pip install requests boto3 pyyaml
pip freeze > requirements.txt      # lưu danh sách packages
pip install -r requirements.txt    # cài lại từ file

---

Foundation — Syntax nhanh cho DevOps

Variables, String, f-string

#!/usr/bin/env python3

# Variables
server_name = "web-prod-01"
port = 8080
is_running = True

# f-string (Python 3.6+) — dùng nhiều trong DevOps scripts
print(f"Server {server_name} listening on port {port}")
host = f"{server_name}.internal.company.com"

# String methods hay dùng
service = "  nginx  "
service.strip()          # → "nginx"
service.upper()          # → "NGINX"
"nginx".startswith("ng") # → True
"192.168.1.1".split(".")  # → ['192', '168', '1', '1']
",".join(["a", "b", "c"]) # → "a,b,c"

List, Dict — cấu trúc dữ liệu cốt lõi

# List — thứ tự quan trọng
servers = ["web-01", "web-02", "db-01"]
servers.append("cache-01")
servers.remove("web-02")

for server in servers:
    print(f"Processing: {server}")

# Dict — key-value, dùng rất nhiều cho config
config = {
    "host": "db-prod.internal",
    "port": 5432,
    "database": "myapp",
    "pool_size": 10
}

# Truy cập an toàn
host = config.get("host", "localhost")  # default nếu key không tồn tại
port = config["port"]

# Dict comprehension
env_vars = {k: v for k, v in config.items() if v is not None}

Error Handling — bắt buộc trong script tự động

import sys
import logging

# Setup logging (quan trọng hơn print cho production script)
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger(__name__)

def connect_to_db(host, port):
    try:
        # ... connection code ...
        logger.info(f"Connected to {host}:{port}")
    except ConnectionRefusedError:
        logger.error(f"Cannot connect to {host}:{port}")
        sys.exit(1)    # exit với error code (quan trọng cho cron/CI)
    except Exception as e:
        logger.error(f"Unexpected error: {e}", exc_info=True)
        raise          # re-raise để caller biết

# Context manager (with) — tự động close file/connection
with open("/etc/app/config.json") as f:
    content = f.read()    # f.close() được gọi tự động dù có exception

---

Module os — Tương tác với hệ thống

import os
import os.path

# === Environment variables ===
db_host = os.environ.get("DB_HOST", "localhost")      # lấy, có default
api_key = os.environ["API_KEY"]                        # lấy, raise nếu không có
os.environ["TEMP_VAR"] = "value"                       # set (chỉ trong process)

# === Path operations ===
current_dir = os.getcwd()                              # thư mục hiện tại
home = os.path.expanduser("~")                         # /home/sontn
config_file = os.path.join(home, ".config", "app.yml") # build path an toàn
os.path.exists("/etc/nginx/nginx.conf")                # True/False
os.path.isfile("/etc/nginx/nginx.conf")                # True nếu là file
os.path.isdir("/etc/nginx")                            # True nếu là thư mục
os.path.basename("/etc/nginx/nginx.conf")              # "nginx.conf"
os.path.dirname("/etc/nginx/nginx.conf")               # "/etc/nginx"

# === File/Directory operations ===
os.makedirs("/opt/app/logs", exist_ok=True)            # tạo cả cây thư mục
os.listdir("/etc/nginx")                               # danh sách file
os.rename("/tmp/file.txt", "/opt/file.txt")            # đổi tên/di chuyển
os.remove("/tmp/file.txt")                             # xoá file
os.rmdir("/tmp/empty_dir")                             # xoá dir rỗng

# Walk qua cây thư mục
for root, dirs, files in os.walk("/var/log/nginx"):
    for filename in files:
        filepath = os.path.join(root, filename)
        size = os.path.getsize(filepath)
        print(f"{filepath}: {size} bytes")

---

Module subprocess — Chạy lệnh shell

Đây là cách để Python gọi các lệnh Linux — thay thế cho Bash:

import subprocess

# === Cách đơn giản — chỉ cần kết quả ===
result = subprocess.run(
    ["systemctl", "is-active", "nginx"],
    capture_output=True,    # bắt stdout + stderr
    text=True               # trả về string (không phải bytes)
)

print(result.stdout)        # "active\n"
print(result.returncode)    # 0 = success, non-0 = failure

# Kiểm tra status dịch vụ
def is_service_running(service_name):
    result = subprocess.run(
        ["systemctl", "is-active", service_name],
        capture_output=True, text=True
    )
    return result.returncode == 0

# === check=True — raise exception nếu lỗi ===
try:
    subprocess.run(
        ["git", "pull", "--ff-only"],
        check=True,
        cwd="/opt/myapp",
        capture_output=True,
        text=True
    )
    print("Git pull successful")
except subprocess.CalledProcessError as e:
    print(f"Git pull failed: {e.stderr}")
    raise

# === Chạy nhiều lệnh, lấy output ===
def get_disk_usage(path="/"):
    result = subprocess.run(
        ["df", "-h", path],
        capture_output=True, text=True, check=True
    )
    # parse output
    lines = result.stdout.strip().split("\n")
    return lines[-1].split()

# === Pipeline: tương đương cmd1 | cmd2 ===
ps = subprocess.Popen(["ps", "aux"], stdout=subprocess.PIPE)
grep = subprocess.Popen(["grep", "nginx"], stdin=ps.stdout,
                        stdout=subprocess.PIPE, text=True)
ps.stdout.close()
output, _ = grep.communicate()
print(output)

Tránh shell=True

subprocess.run("rm -rf " + user_input, shell=True) → nguy hiểm — command injection!
Luôn dùng list arguments và shell=False (mặc định) khi input có thể đến từ bên ngoài.

---

Xử lý File phổ biến trong DevOps

JSON

import json

# Đọc JSON config
with open("/etc/app/config.json") as f:
    config = json.load(f)

db_host = config["database"]["host"]

# Ghi JSON
config["last_updated"] = "2026-01-15"
with open("/etc/app/config.json", "w") as f:
    json.dump(config, f, indent=2)

# Parse JSON từ API response
import urllib.request
with urllib.request.urlopen("http://localhost:8080/health") as response:
    data = json.loads(response.read())
    print(data["status"])

YAML

import yaml   # pip install pyyaml

# Đọc Docker Compose / Kubernetes manifest
with open("docker-compose.yml") as f:
    compose = yaml.safe_load(f)   # safe_load an toàn hơn load

for service_name, service_config in compose["services"].items():
    image = service_config.get("image", "N/A")
    print(f"{service_name}: {image}")

# Cập nhật version trong manifest
compose["services"]["app"]["image"] = "myapp:v2.0.0"
with open("docker-compose.yml", "w") as f:
    yaml.dump(compose, f, default_flow_style=False)

ConfigParser — file .ini / .conf

import configparser

config = configparser.ConfigParser()
config.read("/etc/myapp/settings.ini")

db_host = config["database"]["host"]
db_port = config.getint("database", "port")  # convert type tự động
debug = config.getboolean("app", "debug")

# Ghi config
config["new_section"] = {"key": "value"}
with open("/etc/myapp/settings.ini", "w") as f:
    config.write(f)

---

requests — HTTP API calls

import requests   # pip install requests

# GET request
response = requests.get(
    "https://api.github.com/repos/owner/repo/releases/latest",
    headers={"Authorization": f"token {os.environ['GITHUB_TOKEN']}"},
    timeout=10  # timeout bắt buộc trong script tự động
)
response.raise_for_status()  # raise nếu HTTP 4xx/5xx
latest_version = response.json()["tag_name"]

# POST request — trigger CI/CD
payload = {
    "ref": "main",
    "inputs": {"environment": "production"}
}
response = requests.post(
    "https://api.github.com/repos/owner/repo/actions/workflows/deploy.yml/dispatches",
    headers={
        "Authorization": f"Bearer {os.environ['GITHUB_TOKEN']}",
        "Accept": "application/vnd.github+json"
    },
    json=payload,
    timeout=30
)
response.raise_for_status()
print("Deployment triggered!")

# Health check với retry
def wait_for_service(url, max_retries=10, delay=5):
    import time
    for attempt in range(max_retries):
        try:
            r = requests.get(url, timeout=5)
            if r.status_code == 200:
                print(f"Service ready after {attempt * delay}s")
                return True
        except requests.ConnectionError:
            pass
        print(f"Attempt {attempt + 1}/{max_retries} failed, retrying...")
        time.sleep(delay)
    return False

---

boto3 — AWS SDK Python

import boto3   # pip install boto3

# Authenticate via environment vars hoặc IAM role (best practice)
# export AWS_ACCESS_KEY_ID=...
# export AWS_SECRET_ACCESS_KEY=...
# export AWS_DEFAULT_REGION=ap-southeast-1

# === S3 ===
s3 = boto3.client("s3")

# Upload file lên S3
s3.upload_file("/var/backup/db.sql.gz", "my-bucket", "backups/db.sql.gz")

# Download file
s3.download_file("my-bucket", "config/app.yml", "/tmp/app.yml")

# Liệt kê objects
response = s3.list_objects_v2(Bucket="my-bucket", Prefix="backups/")
for obj in response.get("Contents", []):
    print(f"{obj['Key']}: {obj['Size']} bytes, {obj['LastModified']}")

# === EC2 ===
ec2 = boto3.resource("ec2", region_name="ap-southeast-1")

# Lấy danh sách instances đang chạy
running_instances = ec2.instances.filter(
    Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
for instance in running_instances:
    name = next((tag["Value"] for tag in (instance.tags or [])
                 if tag["Key"] == "Name"), "N/A")
    print(f"{instance.id} | {instance.instance_type} | {name}")

# === Systems Manager — chạy lệnh trên EC2 không cần SSH ===
ssm = boto3.client("ssm")
response = ssm.send_command(
    InstanceIds=["i-1234567890abcdef0"],
    DocumentName="AWS-RunShellScript",
    Parameters={"commands": ["systemctl status nginx"]}
)
command_id = response["Command"]["CommandId"]

---

Script thực tế — Deployment Checker

#!/usr/bin/env python3
"""
Kiểm tra tất cả services sau khi deployment
Dùng trong CI/CD pipeline hoặc cron job
"""
import sys
import time
import logging
import subprocess
import requests
from dataclasses import dataclass

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(message)s"
)
logger = logging.getLogger(__name__)

@dataclass
class ServiceCheck:
    name: str
    url: str = None
    systemd_service: str = None
    expected_http_code: int = 200

CHECKS = [
    ServiceCheck("nginx", systemd_service="nginx"),
    ServiceCheck("API", url="http://localhost:8080/health"),
    ServiceCheck("Database", url="http://localhost:8080/db/health"),
]

def check_systemd(service: str) -> bool:
    result = subprocess.run(
        ["systemctl", "is-active", "--quiet", service]
    )
    return result.returncode == 0

def check_http(url: str, expected_code: int = 200) -> bool:
    try:
        response = requests.get(url, timeout=5)
        return response.status_code == expected_code
    except Exception:
        return False

def main():
    failures = []

    for check in CHECKS:
        ok = True
        if check.systemd_service:
            ok = check_systemd(check.systemd_service)
        elif check.url:
            ok = check_http(check.url, check.expected_http_code)

        status = "✅" if ok else "❌"
        logger.info(f"{status} {check.name}")

        if not ok:
            failures.append(check.name)

    if failures:
        logger.error(f"Failed checks: {', '.join(failures)}")
        sys.exit(1)  # exit code 1 → CI/CD pipeline sẽ fail

    logger.info("All checks passed!")
    sys.exit(0)

if __name__ == "__main__":
    main()

---

Packaging — Cấu trúc Script chuyên nghiệp

devops-scripts/
├── scripts/
│   ├── backup.py
│   ├── health_check.py
│   └── deploy.py
├── lib/
│   ├── aws_utils.py          # boto3 helpers
│   ├── slack_notify.py       # Slack webhook
│   └── config.py             # load config
├── tests/
│   ├── test_backup.py
│   └── test_health.py
├── requirements.txt
├── requirements-dev.txt      # pytest, black, mypy
└── README.md

# Chạy tests
pip install pytest
pytest tests/ -v

# Format code
pip install black
black scripts/

# Type checking
pip install mypy
mypy scripts/backup.py

---

Bảng tổng kết — Modules hay dùng

Module	Dùng cho	Cài đặt
os, os.path	File system, env vars	Built-in
subprocess	Chạy lệnh shell	Built-in
json	Parse/ghi JSON	Built-in
pathlib	Path operations (OOP)	Built-in
logging	Logging chuyên nghiệp	Built-in
argparse	CLI arguments	Built-in
dataclasses	Data objects	Built-in
requests	HTTP API calls	pip install requests
pyyaml	YAML files	pip install pyyaml
boto3	AWS SDK	pip install boto3
paramiko	SSH client	pip install paramiko
click	CLI framework	pip install click
ansible	Configuration mgmt	pip install ansible