Loading...
Devops

Python commands for DevOps Phase 2

1. Creating a Docker Image with Python
 Using docker library to build an image:
python
import docker
client = docker.from_env()
# Dockerfile content
dockerfile_content = “””
FROM python:3.9-slim
WORKDIR /app
COPY . /app
RUN pip install -r requirements.txt
CMD [“python”, “app.py”]

“””
# Create a Docker image
image, build_logs = client.images.build(fileobj=dockerfile_content.encode(‘utf- 8′), tag=’my-python-app’)
for line in build_logs:
print(line)

2. Using psutil for System Monitoring
 Retrieve system metrics such as CPU and memory usage:
python
import psutil
print(“CPU Usage:”, psutil.cpu_percent(interval=1), “%”)
print(“Memory Usage:”, psutil.virtual_memory().percent, “%”)

3. Database Migration with Alembic
 Script to initialize Alembic migrations:
python
from alembic import command
from alembic import config
alembic_cfg = config.Config(“alembic.ini”)
command.upgrade(alembic_cfg, “head”)

4. Using paramiko for SSH Connections
 Execute commands on a remote server via SSH:
python
import paramiko
client = paramiko.SSHClient()
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
client.connect(‘hostname’, username=’user’, password=’your_password’)
stdin, stdout, stderr = client.exec_command(‘ls -la’)
print(stdout.read().decode())
client.close()

5. CloudFormation Stack Creation with boto3
 Creating an AWS CloudFormation stack:
python
import boto3
cloudformation = boto3.client(‘cloudformation’)
with open(‘template.yaml’, ‘r’) as template_file:
template_body = template_file.read()
response = cloudformation.create_stack(
StackName=’MyStack’,
TemplateBody=template_body,
Parameters=[
{
‘ParameterKey’: ‘InstanceType’,
‘ParameterValue’: ‘t2.micro’
},
],
TimeoutInMinutes=5,
Capabilities=[‘CAPABILITY_NAMED_IAM’],
)
print(response)

6. Automating EC2 Instance Management
 Starting and stopping EC2 instances:
python
import boto3
ec2 = boto3.resource(‘ec2’)
# Start an instance
instance = ec2.Instance(‘instance_id’)
instance.start()
# Stop an instance
instance.stop()

7. Automated Backup with shutil
 Backup files to a specific directory:
python
import shutil
import os
source_dir = ‘/path/to/source’
backup_dir = ‘/path/to/backup’
shutil.copytree(source_dir, backup_dir)

8. Using watchdog for File System Monitoring
 Monitor changes in a directory:
python
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class MyHandler(FileSystemEventHandler):
def on_modified(self, event):
print(f’File modified: {event.src_path}’)
event_handler = MyHandler()
observer = Observer()
observer.schedule(event_handler, path=’path/to/monitor’, recursive=False)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()

9. Load Testing with locust
 Basic Locust load testing setup:
python
from locust import HttpUser, task, between
class MyUser(HttpUser):
wait_time = between(1, 3)
@task
def load_test(self):
self.client.get(‘/’)
# To run, save this as locustfile.py and run: locust

10. Integrating with GitHub API
 Fetching repository details using GitHub API:
python
import requests
url = ‘https://api.github.com/repos/user/repo’
response = requests.get(url, headers={‘Authorization’: ‘token
YOUR_GITHUB_TOKEN’})
repo_info = response.json()
print(repo_info)

11. Managing Kubernetes Resources with kubectl
 Using subprocess to interact with Kubernetes:
python
import subprocess
# Get pods
subprocess.run([‘kubectl’, ‘get’, ‘pods’])
# Apply a configuration
subprocess.run([‘kubectl’, ‘apply’, ‘-f’, ‘deployment.yaml’])

12. Using pytest for CI/CD Testing
 Integrate tests in your CI/CD pipeline:
python
# test_example.py
def test_addition():
assert 1 + 1 == 2
# Run pytest in your CI/CD pipeline
subprocess.run([‘pytest’])

13. Creating a Simple CLI Tool with argparse
 Build a command-line interface:
python
import argparse
parser = argparse.ArgumentParser(description=’Process some integers.’)
parser.add_argument(‘integers’, metavar=’N’, type=int, nargs=’+’, help=’an
integer to be processed’)
parser.add_argument(‘–sum’, dest=’accumulate’, action=’store_const’,
const=sum, default=max, help=’sum the integers (default: find the max)’)
args = parser.parse_args()
print(args.accumulate(args.integers))

14. Using dotenv for Environment Variables
 Load environment variables from a .env file:
python
from dotenv import load_dotenv
import os
load_dotenv()
database_url = os.getenv(‘DATABASE_URL’)
print(database_url)

15. Implementing Web Scraping with BeautifulSoup
 Scraping a web page for data:
python
import requests
from bs4 import BeautifulSoup
response = requests.get(‘http://example.com’)
soup = BeautifulSoup(response.text, ‘html.parser’)
for item in soup.find_all(‘h1’):
print(item.text)

16. Using PyYAML for YAML Configuration Files
 Load and dump YAML files:
python
import yaml
# Load YAML file
with open(‘config.yaml’, ‘r’) as file:
config = yaml.safe_load(file)
print(config)
# Dump to YAML file
with open(‘output.yaml’, ‘w’) as file:
yaml.dump(config, file)

17. Creating a Simple Message Queue with RabbitMQ
 Send and receive messages using pika:
python
import pika
# Sending messages
connection =
pika.BlockingConnection(pika.ConnectionParameters(‘localhost’))
channel = connection.channel()
channel.queue_declare(queue=’hello’)
channel.basic_publish(exchange=”, routing_key=’hello’, body=’Hello World!’)
connection.close()
# Receiving messages
def callback(ch, method, properties, body):
print(“Received:”, body)
connection =
pika.BlockingConnection(pika.ConnectionParameters(‘localhost’))
channel = connection.channel()
channel.queue_declare(queue=’hello’)
channel.basic_consume(queue=’hello’, on_message_callback=callback,
auto_ack=True)
channel.start_consuming()

118. Using sentry_sdk for Monitoring
 Integrate Sentry for error tracking:
python
import sentry_sdk
sentry_sdk.init(“YOUR_SENTRY_DSN”)
try:
# Your code that may throw an exception
1 / 0
except Exception as e:
sentry_sdk.capture_exception(e)

19. Using openpyxl for Excel File Manipulation
 Read and write Excel files:
python
from openpyxl import Workbook, load_workbook
# Create a new workbook
wb = Workbook()
ws = wb.active
ws[‘A1’] = ‘Hello’
wb.save(‘sample.xlsx’)
# Load an existing workbook
wb = load_workbook(‘sample.xlsx’)
ws = wb.active
print(ws[‘A1’].value)

20. Using sqlalchemy for Database Interaction
 Define a model and perform CRUD operations:
python
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class User(Base):
__tablename__ = ‘users’
id = Column(Integer, primary_key=True)
name = Column(String)
engine = create_engine(‘sqlite:///example.db’)
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)
session = Session()
# Create
new_user = User(name=’Alice’)
session.add(new_user)

21. Monitoring Docker Containers with docker-py
 Fetch and print the status of running containers:
python
import docker
client = docker.from_env()
containers = client.containers.list()
for container in containers:
print(f’Container Name: {container.name}, Status: {container.status}’)

22. Using flask to Create a Simple API
 Basic API setup with Flask:
python
from flask import Flask, jsonify
app = Flask(__name__)
@app.route(‘/api/data’, methods=[‘GET’])
def get_data():
return jsonify({“message”: “Hello, World!”})
if __name__ == ‘__main__’:
app.run(debug=True)

23. Automating Certificate Renewal with certbot
 Script to renew Let’s Encrypt certificates:
python
import subprocess
# Renew certificates
subprocess.run([‘certbot’, ‘renew’])

24. Using numpy for Data Analysis
 Performing basic numerical operations:
python
import numpy as np
data = np.array([1, 2, 3, 4, 5])
mean_value = np.mean(data)
print(“Mean Value:”, mean_value)

25. Creating and Sending Emails with smtplib
 Send an email using Python:
python
import smtplib
from email.mime.text import MIMEText
sender = ‘you@example.com’
recipient = ‘recipient@example.com’
msg = MIMEText(‘This is a test email.’)
msg[‘Subject’] = ‘Test Email’
msg[‘From’] = sender
msg[‘To’] = recipient
with smtplib.SMTP(‘smtp.example.com’) as server:
server.login(‘username’, ‘password’)
server.send_message(msg)

26. Using schedule for Task Scheduling
 Schedule tasks at regular intervals:
python
import schedule
import time
def job():
print(“Job is running…”)
schedule.every(10).minutes.do(job)
while True:
schedule.run_pending()
time.sleep(1)

27. Using matplotlib for Data Visualization
 Plotting a simple graph:
python
import matplotlib.pyplot as plt
x = [1, 2, 3, 4, 5]
y = [2, 3, 5, 7, 11]
plt.plot(x, y)
plt.xlabel(‘X-axis’)
plt.ylabel(‘Y-axis’)
plt.title(‘Simple Plot’)
plt.show()

28. Creating a Custom Python Package
 Structure your project as a package:
markdown
my_package/
├── __init__.py
├── module1.py
└── module2.py

 setup.py for packaging:

python
from setuptools import setup, find_packages
setup(
name=’my_package’,
version=’0.1′,
packages=find_packages(),
install_requires=[
‘requests’,
‘flask’
],
)

29. Using pytest for Unit Testing
 Writing a simple unit test:
python
# test_sample.py
def add(a, b):
return a + b
def test_add():
assert add(1, 2) == 3

30. Implementing OAuth with requests- oauthlib
 Authenticate with an API using OAuth:
python
from requests_oauthlib import OAuth1Session
oauth = OAuth1Session(client_key=’YOUR_CLIENT_KEY’,
client_secret=’YOUR_CLIENT_SECRET’)
response = oauth.get(‘https://api.example.com/user’)
print(response.json())

31. Using pandas for Data Manipulation
 Load and manipulate data in a CSV file:
python
import pandas as pd
df = pd.read_csv(‘data.csv’)
print(df.head())
# Filter data
filtered_df = df[df[‘column_name’] > 10]
print(filtered_df)

32. Using requests for HTTP Requests
 Making a GET and POST request:
python
import requests
# GET request
response = requests.get(‘https://api.example.com/data’)
print(response.json())
# POST request
data = {‘key’: ‘value’}
response = requests.post(‘https://api.example.com/data’, json=data)
print(response.json())

33. Creating a Basic Web Server with http.server
 Simple HTTP server to serve files:
python
from http.server import SimpleHTTPRequestHandler, HTTPServer
PORT = 8000
handler = SimpleHTTPRequestHandler
with HTTPServer((”, PORT), handler) as httpd:
print(f’Serving on port {PORT}’)
httpd.serve_forever()

34. Using Flask for Webhooks
 Handling incoming webhook requests:
python
from flask import Flask, request
app = Flask(__name__)
@app.route(‘/webhook’, methods=[‘POST’])
def webhook():
data = request.json
print(data)
return ”, 200
if __name__ == ‘__main__’:
app.run(port=5000)

35. Creating a Bash Script with subprocess
 Run shell commands from Python:
python
import subprocess
subprocess.run([‘echo’, ‘Hello, World!’])

36. Using docker-compose with Python
 Programmatically run Docker Compose commands:
python
import subprocess
subprocess.run([‘docker-compose’, ‘up’, ‘-d’])

37. Using moto for Mocking AWS Services in Tests
 Mocking AWS S3 for unit testing:
python
import boto3
from moto import mock_s3
@mock_s3
def test_s3_upload():
s3 = boto3.client(‘s3′, region_name=’us-east-1′)
s3.create_bucket(Bucket=’my-bucket’)
s3.upload_file(‘file.txt’, ‘my-bucket’, ‘file.txt’)
# Test logic here

38. Using asyncio for Asynchronous Tasks
 Run multiple tasks concurrently:
python
import asyncio
async def say_hello():
print(“Hello”)
await asyncio.sleep(1)
print(“World”)

async def main():
await asyncio.gather(say_hello(), say_hello())
asyncio.run(main())

39. Using flask-cors for Cross-Origin Resource Sharing
 Allow CORS in a Flask app:
python
from flask import Flask
from flask_cors import CORS
app = Flask(__name__)
CORS(app)
@app.route(‘/data’, methods=[‘GET’])
def data():
return {“message”: “Hello from CORS!”}
if __name__ == ‘__main__’:
app.run()

40. Using pytest Fixtures for Setup and Teardown
 Create a fixture to manage resources:
python
import pytest
@pytest.fixture
def sample_data():
data = {“key”: “value”}
yield data # This is the test data
# Teardown code here (if necessary)
def test_sample_data(sample_data):
assert sample_data[‘key’] == ‘value’

41. Using http.client for Low-Level HTTP Requests
 Make a raw HTTP GET request:
python
import http.client
conn = http.client.HTTPSConnection(“www.example.com”)
conn.request(“GET”, “/”)
response = conn.getresponse()
print(response.status, response.reason)
data = response.read()
conn.close()

42. Implementing Redis Caching with redis-py
 Basic operations with Redis:
python
import redis
r = redis.StrictRedis(host=’localhost’, port=6379, db=0)
# Set and get value
r.set(‘key’, ‘value’)
print(r.get(‘key’).decode(‘utf-8’))

43. Using json for Data Serialization
 Convert Python objects to JSON:
python
import json
data = {“key”: “value”}
json_data = json.dumps(data)
print(json_data)

44. Using xml.etree.ElementTree for XML Processing
 Parse an XML file:
python
import xml.etree.ElementTree as ET
tree = ET.parse(‘data.xml’)
root = tree.getroot()
for child in root:
print(child.tag, child.attrib)

45. Creating a Virtual Environment with venv
 Programmatically create a virtual environment:
python
import venv
venv.create(‘myenv’, with_pip=True)

46. Using psutil for System Monitoring
 Get system memory usage:
python
import psutil
memory = psutil.virtual_memory()
print(f’Total Memory: {memory.total}, Available Memory:
{memory.available}’)

47. Using sqlite3 for Lightweight Database Management
 Basic SQLite operations:
python
import sqlite3
conn = sqlite3.connect(‘example.db’)
c = conn.cursor()
c.execute(”’CREATE TABLE IF NOT EXISTS users (id INTEGER
PRIMARY KEY, name TEXT)”’)
c.execute(“INSERT INTO users (name) VALUES (‘Alice’)”)
conn.commit()
for row in c.execute(‘SELECT * FROM users’):
print(row)
conn.close()

48. Using pytest to Run Tests in Parallel
 Run tests concurrently:
bash
pytest -n 4 # Run tests in parallel with 4 workers

49. Using argparse for Command-Line Arguments
 Parse command-line arguments:
python
import argparse
parser = argparse.ArgumentParser(description=’Process some integers.’)
parser.add_argument(‘integers’, metavar=’N’, type=int, nargs=’+’,
help=’an integer for the accumulator’)
parser.add_argument(‘–sum’, dest=’accumulate’, action=’store_const’,
const=sum, default=max,
help=’sum the integers (default: find the max)’)
args = parser.parse_args()
print(args.accumulate(args.integers))

50. Using jsonschema for JSON Validation
 Validate JSON against a schema:
python
from jsonschema import validate
from jsonschema.exceptions import ValidationError
schema = {
“type”: “object”,
“properties”: {
“name”: {“type”: “string”},
“age”: {“type”: “integer”, “minimum”: 0}
},
“required”: [“name”, “age”]
}
data = {“name”: “John”, “age”: 30}
try:
validate(instance=data, schema=schema)
print(“Data is valid”)
except ValidationError as e:
print(f”Data is invalid: {e.message}”)

Leave a Reply

Your email address will not be published. Required fields are marked *