Пригождался ли Python по DevOps части

Python в DevOps

Python — один из самых важных языков в DevOps экосистеме. Он активно используется для автоматизации и управления инфраструктурой.

1. Scripting и Automation

import subprocess
import os

class Deployment:
    def __init__(self, app_name, version):
        self.app_name = app_name
        self.version = version
    
    def pull_code(self):
        subprocess.run(["git", "clone", f"https://github.com/company/{self.app_name}"])
    
    def install_dependencies(self):
        subprocess.run(["pip", "install", "-r", "requirements.txt"])
    
    def run_migrations(self):
        subprocess.run(["python", "manage.py", "migrate"])
    
    def deploy(self):
        self.pull_code()
        self.install_dependencies()
        self.run_migrations()
        print(f"Deployed {self.app_name}:{self.version}")

2. Infrastructure as Code (Pulumi)

import pulumi
import pulumi_aws as aws

vpc = aws.ec2.Vpc("main", cidr_block="10.0.0.0/16")
subnet = aws.ec2.Subnet("main", vpc_id=vpc.id, cidr_block="10.0.1.0/24")
instance = aws.ec2.Instance("web",
    ami="ami-0c55b159cbfafe1f0",
    instance_type="t2.micro",
    subnet_id=subnet.id)

pulumi.export("instance_ip", instance.public_ip_address)

3. Мониторинг

import requests
import time
from datetime import datetime

class ServiceMonitor:
    def __init__(self, url, max_response_time=2):
        self.url = url
        self.max_response_time = max_response_time
    
    def check_health(self):
        try:
            start = time.time()
            response = requests.get(f"{self.url}/health", timeout=5)
            elapsed = time.time() - start
            
            if response.status_code != 200:
                self.alert(f"Health check failed: {response.status_code}")
                return False
            
            if elapsed > self.max_response_time:
                self.alert(f"Slow response: {elapsed:.2f}s")
                return False
            
            return True
        except requests.RequestException as e:
            self.alert(f"Connection error: {e}")
            return False
    
    def alert(self, message):
        timestamp = datetime.now().isoformat()
        print(f"[{timestamp}] ALERT: {message}")

monitor = ServiceMonitor("http://localhost:8000")
monitor.check_health()

4. Docker Management

import docker

client = docker.from_env()

container = client.containers.run(
    "python:3.9",
    "python -c 'print(\"Hello\")'",
    detach=False
)

for container in client.containers.list():
    print(f"Container: {container.name}, Status: {container.status}")

5. Kubernetes

from kubernetes import client, config

config.load_kube_config()
v1 = client.CoreV1Api()

pods = v1.list_pod_for_all_namespaces(watch=False)
for pod in pods.items:
    print(f"Pod: {pod.metadata.name} in namespace {pod.metadata.namespace}")

6. Database Backup

import subprocess
from datetime import datetime

class DatabaseBackup:
    def __init__(self, db_name, db_user, backup_dir):
        self.db_name = db_name
        self.db_user = db_user
        self.backup_dir = backup_dir
    
    def backup(self):
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        backup_file = f"{self.backup_dir}/backup_{timestamp}.sql"
        
        cmd = ["pg_dump", f"--username={self.db_user}",
               f"--file={backup_file}", self.db_name]
        
        subprocess.run(cmd)
        print(f"Backup created: {backup_file}")
        return backup_file

backup = DatabaseBackup("production", "postgres", "/backups")
backup.backup()

7. AWS Management

import boto3

ec2 = boto3.resource("ec2", region_name="us-east-1")
instances = ec2.instances.filter(Filters=[{"Name": "instance-state-name", "Values": ["running"]}])

for instance in instances:
    print(f"Instance: {instance.id}, Type: {instance.instance_type}")

s3 = boto3.client("s3")
s3.upload_file("local_file.txt", "my-bucket", "remote_file.txt")

rds = boto3.client("rds")
databases = rds.describe_db_instances()
for db in databases["DBInstances"]:
    print(f"Database: {db['DBInstanceIdentifier']}")

8. CI/CD Pipeline

import requests

class JenkinsPipeline:
    def __init__(self, jenkins_url, job_name):
        self.jenkins_url = jenkins_url
        self.job_name = job_name
    
    def trigger_build(self, parameters=None):
        url = f"{self.jenkins_url}/job/{self.job_name}/buildWithParameters"
        response = requests.post(url, json=parameters)
        
        if response.status_code == 201:
            print(f"Build triggered for {self.job_name}")
        else:
            print(f"Failed: {response.text}")
    
    def get_build_status(self, build_number):
        url = f"{self.jenkins_url}/job/{self.job_name}/{build_number}/api/json"
        response = requests.get(url)
        data = response.json()
        return data.get("result")

jenkins = JenkinsPipeline("http://localhost:8080", "my-app")
jenkins.trigger_build({"VERSION": "1.0.0"})
status = jenkins.get_build_status(42)
print(f"Build status: {status}")

9. Configuration Management (Ansible)

---
- name: Configure Web Servers
  hosts: webservers
  tasks:
    - name: Install Docker
      package:
        name: docker.io
        state: present
    
    - name: Start Docker service
      service:
        name: docker
        state: started
        enabled: yes
    
    - name: Run container
      docker_container:
        name: myapp
        image: myapp:latest
        ports:
          - "8000:8000"
        state: started

DevOps Tools на Python

Тул           | Применение
Ansible       | Configuration Management
Pulumi        | IaC на Python
Fabric        | Remote execution
Boto3         | AWS Management
DockerPy      | Docker Management
Kubernetes    | K8s Management
Celery        | Task Scheduling
APScheduler   | Job Scheduling

Best Practices

class Infrastructure:
    def __init__(self, env):
        self.env = env
        self.config = self.load_config()
    
    def load_config(self):
        import os
        return {
            "aws_region": os.getenv("AWS_REGION", "us-east-1"),
            "app_name": os.getenv("APP_NAME")
        }
    
    def deploy(self):
        try:
            self.validate_config()
            self.backup_current_state()
            self.apply_changes()
            self.verify_deployment()
            self.cleanup()
        except Exception as e:
            self.rollback()
            raise

Заключение

Python в DevOps:

• Автоматизация инфраструктуры
• Infrastructure as Code (IaC)
• Управление облаком (AWS, GCP, Azure)
• Мониторинг и логирование
• CI/CD пайплайны
• Контейнеризация (Docker, Kubernetes)

Выбирают Python за: простоту, мощную экосистему, скорость разработки, кроссплатформенность.