1. File Operations
Read a file:
python
with open(‘file.txt’, ‘r’) as file:
content = file.read()
print(content)
Write to a file:
python
with open(‘output.txt’, ‘w’) as file:
file.write(‘Hello, DevOps!’)
2. Environment Variables
Get an environment variable:
python
import os
db_user = os.getenv(‘DB_USER’)
print(db_user)
Set an environment variable:
python
import os
os.environ[‘NEW_VAR’] = ‘value’
3. Subprocess Management
Run shell commands:
python
import subprocess
result = subprocess.run([‘ls’, ‘-l’], capture_output=True, text=True)
print(result.stdout)
4. API Requests
Make a GET request:
python
import requests
response = requests.get(‘https://api.example.com/data’)
print(response.json())
5. JSON Handling
Read JSON from a file:
python
import json
with open(‘data.json’, ‘r’) as file:
data = json.load(file)
print(data)
Write JSON to a file:
python
import json
data = {‘name’: ‘DevOps’, ‘type’: ‘Workflow’}
with open(‘output.json’, ‘w’) as file:
json.dump(data, file, indent=4)
6. Logging
Basic logging setup:
python
import logging
logging.basicConfig(level=logging.INFO)
logging.info(‘This is an informational message’)
7. Working with Databases
Connect to a SQLite database:
python
import sqlite3
conn = sqlite3.connect(‘example.db’)
cursor = conn.cursor()
cursor.execute(‘CREATE TABLE IF NOT EXISTS users (id INTEGER
PRIMARY KEY, name TEXT)’)
conn.commit()
conn.close()
8. Automation with Libraries
Using Paramiko for SSH connections:
python
import paramiko
ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh.connect(‘hostname’, username=’user’, password=’password’)
stdin, stdout, stderr = ssh.exec_command(‘ls’)
print(stdout.read().decode())
ssh.close()
9. Error Handling
Try-except block:
python
try:
# code that may raise an exception
risky_code()
except Exception as e:
print(f’Error occurred: {e}’)
10. Docker Integration
Using the docker package to interact with Docker:
python
import docker
client = docker.from_env()
containers = client.containers.list()
for container in containers:
print(container.name)
11. Working with YAML Files
Read a YAML file:
python
import yaml
with open(‘config.yaml’, ‘r’) as file:
config = yaml.safe_load(file)
print(config)
Write to a YAML file:
python
import yaml
data = {‘name’: ‘DevOps’, ‘version’: ‘1.0’}
with open(‘output.yaml’, ‘w’) as file:
yaml.dump(data, file)
12. Parsing Command-Line Arguments
Using argparse:
python
import argparse
parser = argparse.ArgumentParser(description=’Process some integers.’)
parser.add_argument(‘–num’, type=int, help=’an integer for the accumulator’)
args = parser.parse_args()
print(args.num)
13. Monitoring System Resources
Using psutil to monitor system resources:
python
import psutil
print(f”CPU Usage: {psutil.cpu_percent()}%”)
print(f”Memory Usage: {psutil.virtual_memory().percent}%”)
14. Handling HTTP Requests with Flask
Basic Flask API:
python
from flask import Flask, jsonify
app = Flask(__name__)
@app.route(‘/health’, methods=[‘GET’])
def health_check():
return jsonify({‘status’: ‘healthy’})
if __name__ == ‘__main__’:
app.run(host=’0.0.0.0′, port=5000)
15. Creating Docker Containers
Using the Docker SDK to create a container:
python
import docker
client = docker.from_env()
container = client.containers.run(‘ubuntu’, ‘echo Hello World’, detach=True)
print(container.logs())
16. Scheduling Tasks
Using schedule for task scheduling:
python
import schedule
import time
def job():
print(“Running scheduled job…”)
schedule.every(1).minutes.do(job)
while True:
schedule.run_pending()
time.sleep(1)
17. Version Control with Git
Using GitPython to interact with Git repositories:
python
import git
repo = git.Repo(‘/path/to/repo’)
repo.git.add(‘file.txt’)
repo.index.commit(‘Added file.txt’)
18. Email Notifications
Sending emails using smtplib:
python
import smtplib
from email.mime.text import MIMEText
msg = MIMEText(‘This is the body of the email’)
msg[‘Subject’] = ‘Email Subject’
msg[‘From’] = ‘you@example.com’
msg[‘To’] = ‘recipient@example.com’
with smtplib.SMTP(‘smtp.example.com’, 587) as server:
server.starttls()
server.login(‘your_username’, ‘your_password’)
server.send_message(msg)
19. Creating Virtual Environments
Creating and activating a virtual environment:
python
import os
import subprocess
# Create virtual environment
subprocess.run([‘python3’, ‘-m’, ‘venv’, ‘myenv’])
# Activate virtual environment (Windows)
os.system(‘myenv\\Scripts\\activate’)
# Activate virtual environment (Linux/Mac)
os.system(‘source myenv/bin/activate’)
20. Integrating with CI/CD Tools
Using the requests library to trigger a Jenkins job:
python
import requests
url = ‘http://your-jenkins-url/job/your-job-name/build’
response = requests.post(url, auth=(‘user’, ‘token’))
print(response.status_code)
21. Database Migration
Using Alembic for database migrations:
bash
alembic revision -m “initial migration”
alembic upgrade head
22. Testing Code
Using unittest for unit testing:
python
import unittest
def add(a, b):
return a + b
class TestMathFunctions(unittest.TestCase):
def test_add(self):
self.assertEqual(add(2, 3), 5)
if __name__ == ‘__main__’:
unittest.main()
23. Data Transformation with Pandas
Using pandas for data manipulation:
python
import pandas as pd
df = pd.read_csv(‘data.csv’)
df[‘new_column’] = df[‘existing_column’] * 2
df.to_csv(‘output.csv’, index=False)
24. Using Python for Infrastructure as Code
Using boto3 for AWS operations:
python
import boto3
ec2 = boto3.resource(‘ec2’)
instances = ec2.instances.filter(Filters=[{‘Name’: ‘instance-state-name’,
‘Values’: [‘running’]}])
for instance in instances:
print(instance.id, instance.state)
25. Web Scraping
Using BeautifulSoup to scrape web pages:
python
import requests
from bs4 import BeautifulSoup
response = requests.get(‘http://example.com’)
soup = BeautifulSoup(response.content, ‘html.parser’)
print(soup.title.string)
26. Using Fabric for Remote Execution
Running commands on a remote server:
python
from fabric import Connection
conn = Connection(host=’user@hostname’, connect_kwargs={‘password’:
‘your_password’})
conn.run(‘uname -s’)
27. Automating AWS S3 Operations
Upload and download files using boto3:
python
import boto3
s3 = boto3.client(‘s3’)
# Upload a file
s3.upload_file(‘local_file.txt’, ‘bucket_name’, ‘s3_file.txt’)
# Download a file
s3.download_file(‘bucket_name’, ‘s3_file.txt’, ‘local_file.txt’)
28. Monitoring Application Logs
Tail logs using tail -f equivalent in Python:
python
import time
def tail_f(file):
file.seek(0, 2) # Move to the end of the file
while True:
line = file.readline()
if not line:
time.sleep(0.1) # Sleep briefly
continue
print(line)
with open(‘app.log’, ‘r’) as log_file:
tail_f(log_file)
29. Container Health Checks
Check the health of a running Docker container:
python
import docker
client = docker.from_env()
container = client.containers.get(‘container_id’)
print(container.attrs[‘State’][‘Health’][‘Status’])
30. Using requests for Rate-Limited APIs
Handle rate limiting in API requests:
python
import requests
import time
url = ‘https://api.example.com/data’
while True:
response = requests.get(url)
if response.status_code == 200:
print(response.json())
break
elif response.status_code == 429: # Too Many Requests
time.sleep(60) # Wait a minute before retrying
else:
print(‘Error:’, response.status_code)
break
31. Docker Compose Integration
Using docker-compose in Python:
python
import os
import subprocess
# Start services defined in docker-compose.yml
subprocess.run([‘docker-compose’, ‘up’, ‘-d’])
# Stop services
subprocess.run([‘docker-compose’, ‘down’])
32. Terraform Execution
Executing Terraform commands with subprocess:
python
import subprocess
# Initialize Terraform
subprocess.run([‘terraform’, ‘init’])
# Apply configuration
subprocess.run([‘terraform’, ‘apply’, ‘-auto-approve’])
33. Working with Prometheus Metrics
Scraping and parsing Prometheus metrics:
python
import requests
response = requests.get(‘http://localhost:9090/metrics’)
metrics = response.text.splitlines()
for metric in metrics:
print(metric)
34. Using pytest for Testing
Simple test case with pytest:
python
def add(a, b):
return a + b
def test_add():
assert add(2, 3) == 5
35. Creating Webhooks
Using Flask to create a simple webhook:
python
from flask import Flask, request
app = Flask(__name__)
@app.route(‘/webhook’, methods=[‘POST’])
def webhook():
data = request.json
print(‘Received data:’, data)
return ‘OK’, 200
if __name__ == ‘__main__’:
app.run(port=5000)
36. Using Jinja2 for Configuration Templates
Render configuration files with Jinja2:
python
from jinja2 import Template
template = Template(‘Hello, {{ name }}!’)
rendered = template.render(name=’DevOps’)
print(rendered)
37. Encrypting and Decrypting Data
Using cryptography to encrypt and decrypt:
python
from cryptography.fernet import Fernet
# Generate a key
key = Fernet.generate_key()
cipher_suite = Fernet(key)
# Encrypt
encrypted_text = cipher_suite.encrypt(b’Secret Data’)
# Decrypt
decrypted_text = cipher_suite.decrypt(encrypted_text)
print(decrypted_text.decode())
38. Error Monitoring with Sentry
Sending error reports to Sentry:
python
import sentry_sdk
sentry_sdk.init(‘your_sentry_dsn’)
def divide(a, b):
return a / b
try:
divide(1, 0)
except ZeroDivisionError as e:
sentry_sdk.capture_exception(e)
39. Setting Up Continuous Integration with GitHub Actions
Sample workflow file (.github/workflows/ci.yml):
yaml
name: CI
on: [push]
jobs:
build:
runs-on: ubuntu-latest
steps:
– uses: actions/checkout@v2
– name: Set up Python
uses: actions/setup-python@v2
with:
python-version: ‘3.8’
– name: Install dependencies
run: |
pip install -r requirements.txt
– name: Run tests
run: |
pytest
40. Creating a Simple API with FastAPI
Using FastAPI for high-performance APIs:
python
from fastapi import FastAPI
app = FastAPI()
@app.get(‘/items/{item_id}’)
async def read_item(item_id: int):
return {‘item_id’: item_id}
if __name__ == ‘__main__’:
import uvicorn
uvicorn.run(app, host=’0.0.0.0′, port=8000)
41. Log Aggregation with ELK Stack
Sending logs to Elasticsearch:
python
from elasticsearch import Elasticsearch
es = Elasticsearch([‘http://localhost:9200’])
log = {‘level’: ‘info’, ‘message’: ‘This is a log message’}
es.index(index=’logs’, body=log)
42. Using pandas for ETL Processes
Performing ETL with pandas:
python
import pandas as pd
# Extract
data = pd.read_csv(‘source.csv’)
# Transform
data[‘new_column’] = data[‘existing_column’].apply(lambda x: x * 2)
# Load
data.to_csv(‘destination.csv’, index=False)
43. Serverless Applications with AWS Lambda
Deploying a simple AWS Lambda function:
python
import json
def lambda_handler(event, context):
return {
‘statusCode’: 200,
‘body’: json.dumps(‘Hello from Lambda!’)
}
44. Working with Redis
Basic operations with Redis using redis-py:
python
import redis
r = redis.StrictRedis(host=’localhost’, port=6379, db=0)
# Set a key
r.set(‘foo’, ‘bar’)
# Get a key
print(r.get(‘foo’))
45. Using pyngrok for Tunneling
Create a tunnel to expose a local server:
python
from pyngrok import ngrok
# Start the tunnel
public_url = ngrok.connect(5000)
print(‘Public URL:’, public_url)
# Keep the tunnel open
input(‘Press Enter to exit…’)
46. Creating a REST API with Flask- RESTful
Building REST APIs with Flask-RESTful:
python
from flask import Flask
from flask_restful import Resource, Api
app = Flask(__name__)
api = Api(app)
class HelloWorld(Resource):
def get(self):
return {‘hello’: ‘world’}
api.add_resource(HelloWorld, ‘/’)
if __name__ == ‘__main__’:
app.run(debug=True)
47. Using asyncio for Asynchronous Tasks
Running asynchronous tasks in Python:
python
import asyncio
async def main():
print(‘Hello’)
await asyncio.sleep(1)
print(‘World’)
asyncio.run(main())
48. Network Monitoring with scapy
Packet sniffing using scapy:
python
from scapy.all import sniff
def packet_callback(packet):
print(packet.summary())
sniff(prn=packet_callback, count=10)
49. Handling Configuration Files with configparser
Reading and writing to INI configuration files:
python
import configparser
config = configparser.ConfigParser()
config.read(‘config.ini’)
print(config[‘DEFAULT’][‘SomeSetting’])
config[‘DEFAULT’][‘NewSetting’] = ‘Value’
with open(‘config.ini’, ‘w’) as configfile:
config.write(configfile)
50. WebSocket Client Example
Creating a WebSocket client with websocket-client:
python
import websocket
def on_message(ws, message):
print(“Received message:”, message)
ws = websocket.WebSocketApp(“ws://echo.websocket.org”,
on_message=on_message)
ws.run_forever()